Ai Matching Records

Build an AI Record Matcher

Match and deduplicate records across datasets with DSPy - blocking to narrow candidates, pairwise LM scoring, and transitive closure to group all matches.

Step 1: Understand the matching task

Ask the user: 1. What records are you matching? (contacts, companies, tickets, products, etc.) 2. Which fields matter? (name, email, phone, address, description, etc.) 3. How many records? (100s vs millions changes blocking strategy significantly) 4. What defines a match? (exact same entity, or "close enough to merge"?) 5. What to do with matches? (deduplicate, merge fields, link IDs, flag for review)

When NOT to use AI matching

• Single-field exact match — if email == email or id == id covers your case, use SQL JOIN or a hash lookup. No LM needed.
• Clean data with unique identifiers — if records already have a shared key (user_id, EIN, ISBN), join on it directly.
• Small datasets where manual review is faster — under 50 records, a human can review pairs in minutes.
• Simple fuzzy string matching covers it — tools like rapidfuzz or fuzzywuzzy handle typos and abbreviations cheaply. Add an LM only when semantic understanding is needed ("IBM" = "International Business Machines").

Step 2: Blocking strategies

Never compare all N×N pairs — that creates O(n²) LM calls. Blocking narrows candidates to a small set of plausible pairs first.

from itertools import combinations
import jellyfish

def block_by_phonetic_name(records):
    """Group records by Soundex of first+last name, return candidate pairs."""
    buckets = {}
    for record in records:
        key = jellyfish.soundex(record["name"].lower())
        buckets.setdefault(key, []).append(record)

    pairs = []
    for bucket in buckets.values():
        if len(bucket) > 1:
            pairs.extend(combinations(bucket, 2))
    return pairs

def block_by_email_domain(records):
    """Block contacts that share email domain — likely same company."""
    buckets = {}
    for record in records:
        domain = record.get("email", "@").split("@")[-1]
        if domain and domain != "gmail.com":  # skip generic domains
            buckets.setdefault(domain, []).append(record)

    pairs = []
    for bucket in buckets.values():
        if len(bucket) > 1:
            pairs.extend(combinations(bucket, 2))
    return pairs

Step 3: Build the pairwise comparison signature

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class CompareRecords(dspy.Signature):
    """Determine whether two records refer to the same real-world entity.
    Consider semantic equivalence - 'IBM' matches 'International Business Machines Corp.'
    Focus on substance, not formatting differences."""
    record_a: str = dspy.InputField(desc="First record as a formatted string of field: value pairs")
    record_b: str = dspy.InputField(desc="Second record as a formatted string of field: value pairs")
    is_match: bool = dspy.OutputField(desc="True if both records refer to the same entity")
    confidence: float = dspy.OutputField(desc="Confidence score between 0.0 and 1.0")
    explanation: str = dspy.OutputField(desc="Brief explanation of why these are or are not the same entity")

matcher = dspy.ChainOfThought(CompareRecords)

Helper to format a record dict as a readable string:

def format_record(record: dict) -> str:
    return "\n".join(f"{k}: {v}" for k, v in record.items() if v)

Step 4: Full matching pipeline

import dspy
from itertools import combinations

class RecordMatcher(dspy.Module):
    def __init__(self, match_threshold=0.6, auto_merge_threshold=0.9):
        self.compare = dspy.ChainOfThought(CompareRecords)
        self.match_threshold = match_threshold
        self.auto_merge_threshold = auto_merge_threshold

    def forward(self, records: list[dict]) -> dict:
        # Phase 1 - blocking: get candidate pairs
        candidate_pairs = self.block(records)

        # Phase 2 - pairwise scoring: score each candidate pair
        scored_pairs = []
        for a, b in candidate_pairs:
            result = self.compare(
                record_a=format_record(a),
                record_b=format_record(b),
            )
            scored_pairs.append({
                "record_a": a,
                "record_b": b,
                "is_match": result.is_match,
                "confidence": result.confidence,
                "explanation": result.explanation,
            })

        # Phase 3 - threshold routing
        auto_merge = [p for p in scored_pairs if p["confidence"] >= self.auto_merge_threshold]
        needs_review = [p for p in scored_pairs if self.match_threshold <= p["confidence"] < self.auto_merge_threshold]
        rejected = [p for p in scored_pairs if p["confidence"] < self.match_threshold]

        # Phase 4 - transitive closure: if A=B and B=C, then A=C
        match_pairs = [(p["record_a"]["id"], p["record_b"]["id"]) for p in auto_merge]
        clusters = self.transitive_closure(match_pairs, records)

        return {
            "clusters": clusters,
            "auto_merge": auto_merge,
            "needs_review": needs_review,
            "rejected": rejected,
        }

    def block(self, records):
        """Override with domain-specific blocking. Default - all pairs (only for small datasets)."""
        return list(combinations(records, 2))

    def transitive_closure(self, match_pairs, records):
        """Union-Find to group all transitively matched records into clusters."""
        parent = {r["id"]: r["id"] for r in records}

        def find(x):
            while parent[x] != x:
                parent[x] = parent[parent[x]]
                x = parent[x]
            return x

        def union(x, y):
            parent[find(x)] = find(y)

        for a_id, b_id in match_pairs:
            union(a_id, b_id)

        clusters = {}
        for record in records:
            root = find(record["id"])
            clusters.setdefault(root, []).append(record)

        return list(clusters.values())

Step 5: Merge strategies

Once you have clusters of matching records, decide how to merge them:

def merge_cluster(cluster: list[dict], strategy="most_complete") -> dict:
    """Merge a cluster of matching records into one canonical record."""
    if strategy == "most_complete":
        # Keep the record with the most non-null fields
        return max(cluster, key=lambda r: sum(1 for v in r.values() if v))

    elif strategy == "newest":
        # Keep the most recently updated record
        return max(cluster, key=lambda r: r.get("updated_at", ""))

    elif strategy == "union":
        # Combine all fields, preferring non-null values from the first record that has them
        merged = {}
        for record in cluster:
            for k, v in record.items():
                if k not in merged or not merged[k]:
                    merged[k] = v
        return merged

    elif strategy == "custom":
        # Per-field rules: prefer email from oldest record, name from most complete, etc.
        raise NotImplementedError("Implement per-field merge logic for your use case")

Step 6: Confidence thresholds

Route pairs based on confidence score - do not require human review for everything:

Tune these thresholds using your labeled pair data (see Step 7).

Step 7: Evaluate and optimize

Label a sample of record pairs as match/no-match to measure precision and recall:

from dspy.evaluate import Evaluate

# Labeled pairs - each example has record_a, record_b, and ground truth is_match
trainset = [
    dspy.Example(
        record_a="name: John Smith\nemail: john@acme.com\nphone: 555-1234",
        record_b="name: Jon Smith\nemail: john.smith@acme.com\nphone: 5551234",
        is_match=True,
        confidence=1.0,
        explanation="Same person, minor formatting differences in name/phone/email"
    ).with_inputs("record_a", "record_b"),
    # Add 20-50+ labeled pairs covering easy matches, near-misses, and clear non-matches
]

devset = trainset[len(trainset)*4//5:]
trainset = trainset[:len(trainset)*4//5]

def match_metric(example, pred, trace=None):
    """Precision-focused metric - penalize false positives more than false negatives."""
    correct_decision = pred.is_match == example.is_match
    if not correct_decision and pred.is_match:
        return 0.0  # false positive - penalize hard
    return float(correct_decision)

evaluator = Evaluate(devset=devset, metric=match_metric, num_threads=4, display_progress=True)
score = evaluator(matcher)
print(f"Baseline: {score}%")

# Optimize with BootstrapFewShot
optimizer = dspy.BootstrapFewShot(metric=match_metric, max_bootstrapped_demos=4)
optimized_matcher = optimizer.compile(matcher, trainset=trainset)

improved = evaluator(optimized_matcher)
print(f"Optimized: {improved}%")

optimized_matcher.save("record_matcher.json")

Key patterns

Asymmetric fields

Some fields are more diagnostic than others. Weight them explicitly in the signature:

class CompareContacts(dspy.Signature):
    """Determine if two contact records refer to the same person.
    Email is the strongest signal. Name variations (nicknames, middle names) are common.
    Phone numbers may be formatted differently but represent the same number."""
    record_a: str = dspy.InputField()
    record_b: str = dspy.InputField()
    is_match: bool = dspy.OutputField()
    confidence: float = dspy.OutputField(desc="0.0 to 1.0")
    explanation: str = dspy.OutputField()

Large-scale matching with embeddings

For datasets too large for phonetic blocking, use embedding similarity as the blocking layer:

import numpy as np

def embed_records(records, embed_fn):
    """Embed each record as a single string for similarity search."""
    texts = [format_record(r) for r in records]
    return np.array([embed_fn(t) for t in texts])

def block_by_embedding(records, embeddings, top_k=5, threshold=0.8):
    """Return pairs whose embeddings exceed the similarity threshold."""
    from sklearn.metrics.pairwise import cosine_similarity
    sim_matrix = cosine_similarity(embeddings)
    pairs = []
    n = len(records)
    for i in range(n):
        for j in range(i+1, n):
            if sim_matrix[i][j] >= threshold:
                pairs.append((records[i], records[j]))
    return pairs

Gotchas

• Skipping blocking and making O(n²) LM calls - always narrow candidates with a cheap heuristic first. Even rough phonetic or token-overlap blocking reduces pairs by 99%+ on typical datasets.
• Using string equality for comparison fields - do not compare fields with == in code before passing to the LM. Let the LM judge semantic equivalence so "IBM" matches "International Business Machines Corp."
• Using `dspy.Assert`/`dspy.Suggest` for output validation - use dspy.Refine with a reward function instead. dspy.Assert raises exceptions on constraint violations; dspy.Refine retries with feedback, which is the right pattern for improving match quality.
• Skipping transitive closure - if A matches B and B matches C, then A, B, and C are the same entity. Without Union-Find or similar, you will merge A+B and B+C separately but miss the A+B+C cluster.
• Outputting only is_match without a confidence score - boolean output makes threshold-based routing (auto-merge vs human review vs reject) impossible. Always include confidence: float in the output signature.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Compare two items with reasoning - see /dspy-chain-of-thought
• Retry with feedback when match quality is low - see /dspy-refine
• Sample multiple match decisions and pick the most consistent - see /dspy-best-of-n
• Generate labeled pair examples when you have none - see /ai-generating-data
• Measure and improve match precision/recall - see /ai-improving-accuracy
• Score similarity instead of binary match/no-match - see /ai-scoring
• Install `/ai-do` if you do not have it - it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• For worked examples (CRM deduplication, company name matching, ticket deduplication), see examples.md

Record Matching Examples

CRM Contact Deduplication

Match people by name, email, and phone across messy imported data:

import dspy
import jellyfish
from itertools import combinations

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class CompareContacts(dspy.Signature):
    """Determine if two CRM contact records refer to the same person.
    Email match is strong evidence. Name variations like 'Bob'/'Robert' are common.
    Phone numbers may differ in formatting but represent the same line."""
    record_a: str = dspy.InputField(desc="First contact as field: value pairs")
    record_b: str = dspy.InputField(desc="Second contact as field: value pairs")
    is_match: bool = dspy.OutputField(desc="True if both contacts are the same person")
    confidence: float = dspy.OutputField(desc="Confidence from 0.0 to 1.0")
    explanation: str = dspy.OutputField(desc="Key reason for match or non-match decision")

compare = dspy.ChainOfThought(CompareContacts)

# Sample CRM data with messy duplicates
contacts = [
    {"id": "1", "name": "Robert Johnson", "email": "rjohnson@acme.com", "phone": "415-555-0101", "company": "Acme Corp"},
    {"id": "2", "name": "Bob Johnson",    "email": "r.johnson@acme.com", "phone": "4155550101",   "company": "ACME"},
    {"id": "3", "name": "Sarah Lee",      "email": "sarah@techco.io",    "phone": "212-555-0199", "company": "TechCo"},
    {"id": "4", "name": "Sara Lee",       "email": "slee@techco.io",     "phone": "212-555-0199", "company": "TechCo Inc"},
    {"id": "5", "name": "James Park",     "email": "jpark@startup.com",  "phone": "",             "company": "Startup"},
]

def format_contact(c):
    return "\n".join(f"{k}: {v}" for k, v in c.items() if k != "id" and v)

def block_contacts(contacts):
    """Block by Soundex of last name to reduce pairs."""
    buckets = {}
    for c in contacts:
        last = c["name"].split()[-1] if c["name"] else ""
        key = jellyfish.soundex(last.lower())
        buckets.setdefault(key, []).append(c)
    pairs = []
    for bucket in buckets.values():
        if len(bucket) > 1:
            pairs.extend(combinations(bucket, 2))
    return pairs

# Block then score
candidate_pairs = block_contacts(contacts)
print(f"Comparing {len(candidate_pairs)} candidate pairs (down from {len(contacts)*(len(contacts)-1)//2} total)")

results = []
for a, b in candidate_pairs:
    result = compare(record_a=format_contact(a), record_b=format_contact(b))
    results.append({
        "ids": (a["id"], b["id"]),
        "names": (a["name"], b["name"]),
        "is_match": result.is_match,
        "confidence": result.confidence,
        "explanation": result.explanation,
    })
    print(f"  {a['name']} vs {b['name']} - match={result.is_match} ({result.confidence:.2f}): {result.explanation}")

# Route by confidence
auto_merge   = [r for r in results if r["confidence"] >= 0.9]
needs_review = [r for r in results if 0.6 <= r["confidence"] < 0.9]
rejected     = [r for r in results if r["confidence"] < 0.6]

print(f"\nAuto-merge: {len(auto_merge)}, Needs review: {len(needs_review)}, Rejected: {len(rejected)}")

Expected output:

Comparing 2 candidate pairs (down from 10 total)
  Robert Johnson vs Bob Johnson - match=True (0.92): Same email domain and phone number, common nickname Robert/Bob
  Sarah Lee vs Sara Lee - match=True (0.88): Same phone and company, name is a common spelling variant

Company Name Matching

Match legal entity names to canonical records, handling abbreviations, suffixes, and acronyms:

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class MatchCompanyName(dspy.Signature):
    """Determine whether two company name strings refer to the same legal entity.
    Common variations - abbreviations (IBM/International Business Machines),
    legal suffixes (Corp/Corporation/Inc/Ltd), punctuation, and parent/subsidiary names."""
    name_a: str = dspy.InputField(desc="First company name string")
    name_b: str = dspy.InputField(desc="Second company name string")
    is_match: bool = dspy.OutputField(desc="True if both names refer to the same company")
    confidence: float = dspy.OutputField(desc="Confidence from 0.0 to 1.0")
    explanation: str = dspy.OutputField()

company_matcher = dspy.ChainOfThought(MatchCompanyName)

# Test cases spanning easy to hard
test_pairs = [
    ("IBM", "International Business Machines Corp."),
    ("Apple Inc.", "Apple Computer, Inc."),
    ("3M", "Minnesota Mining and Manufacturing Company"),
    ("Goldman Sachs", "Goldman Sachs Group, Inc."),
    ("Amazon", "Amazon Web Services"),         # parent vs subsidiary - tricky
    ("Microsoft", "MicroSoft Corporation"),
    ("Google LLC", "Alphabet Inc."),            # subsidiary vs parent - distinct
]

for name_a, name_b in test_pairs:
    result = company_matcher(name_a=name_a, name_b=name_b)
    verdict = "MATCH" if result.is_match else "DISTINCT"
    print(f"[{verdict} {result.confidence:.2f}] '{name_a}' vs '{name_b}'")
    print(f"  {result.explanation}")

# Optimize with labeled examples
trainset = [
    dspy.Example(
        name_a="IBM",
        name_b="International Business Machines",
        is_match=True, confidence=1.0,
        explanation="IBM is the universally recognized abbreviation for International Business Machines"
    ).with_inputs("name_a", "name_b"),
    dspy.Example(
        name_a="Apple Inc.",
        name_b="Apple Records",
        is_match=False, confidence=0.95,
        explanation="Different companies - Apple Inc. is a tech company, Apple Records is a music label"
    ).with_inputs("name_a", "name_b"),
    dspy.Example(
        name_a="Meta Platforms Inc.",
        name_b="Facebook, Inc.",
        is_match=True, confidence=0.98,
        explanation="Facebook rebranded to Meta Platforms in 2021 - same legal entity"
    ).with_inputs("name_a", "name_b"),
]

def company_metric(example, pred, trace=None):
    return float(pred.is_match == example.is_match)

optimizer = dspy.BootstrapFewShot(metric=company_metric, max_bootstrapped_demos=3)
optimized = optimizer.compile(company_matcher, trainset=trainset)
optimized.save("company_name_matcher.json")

Support Ticket Deduplication

Find duplicate tickets describing the same underlying issue so you do not work the same bug twice:

import dspy
from itertools import combinations

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class CompareTickets(dspy.Signature):
    """Determine whether two support tickets describe the same underlying issue.
    Different users may describe the same bug in completely different words.
    Focus on the root problem, not surface wording or affected user."""
    ticket_a: str = dspy.InputField(desc="First ticket - title and description")
    ticket_b: str = dspy.InputField(desc="Second ticket - title and description")
    is_duplicate: bool = dspy.OutputField(desc="True if both tickets describe the same root issue")
    confidence: float = dspy.OutputField(desc="Confidence from 0.0 to 1.0")
    explanation: str = dspy.OutputField(desc="What makes them the same or different issue")

ticket_deduper = dspy.ChainOfThought(CompareTickets)

# Open tickets to deduplicate
tickets = [
    {
        "id": "T-101",
        "title": "Login page not loading",
        "description": "When I go to app.example.com/login the page spins forever and never loads. Started this morning.",
    },
    {
        "id": "T-102",
        "title": "Cannot access my account",
        "description": "The sign-in screen is broken. Tried Chrome and Firefox - both stuck on loading. Happening since 9am.",
    },
    {
        "id": "T-103",
        "title": "Password reset email not arriving",
        "description": "I requested a password reset 30 minutes ago and the email never came. Checked spam folder.",
    },
    {
        "id": "T-104",
        "title": "Authentication is down",
        "description": "Our whole team cannot log in. The login endpoint appears to be returning 503 errors.",
    },
    {
        "id": "T-105",
        "title": "Forgot password flow broken",
        "description": "Reset password emails are not being sent. Multiple users affected.",
    },
]

def format_ticket(t):
    return f"Title: {t['title']}\nDescription: {t['description']}"

def block_tickets_by_tokens(tickets, min_shared=2):
    """Block tickets sharing at least min_shared title tokens."""
    def tokens(t):
        stopwords = {"the", "a", "an", "is", "not", "my", "i", "and", "or", "to", "in"}
        return set(t["title"].lower().split()) - stopwords

    pairs = []
    for a, b in combinations(tickets, 2):
        if len(tokens(a) & tokens(b)) >= min_shared:
            pairs.append((a, b))
    return pairs

candidate_pairs = block_tickets_by_tokens(tickets)
print(f"Candidate pairs after blocking: {len(candidate_pairs)}")

# Score pairs
matches = []
for a, b in candidate_pairs:
    result = ticket_deduper(
        ticket_a=format_ticket(a),
        ticket_b=format_ticket(b),
    )
    if result.is_duplicate and result.confidence >= 0.7:
        matches.append((a["id"], b["id"], result.confidence, result.explanation))
        print(f"  DUPLICATE ({result.confidence:.2f}): {a['id']} + {b['id']}")
        print(f"    {result.explanation}")

# Transitive closure - group all related tickets
parent = {t["id"]: t["id"] for t in tickets}

def find(x):
    while parent[x] != x:
        parent[x] = parent[parent[x]]
        x = parent[x]
    return x

def union(x, y):
    parent[find(x)] = find(y)

for a_id, b_id, _, _ in matches:
    union(a_id, b_id)

# Print duplicate groups
groups = {}
for t in tickets:
    root = find(t["id"])
    groups.setdefault(root, []).append(t["id"])

print("\nDuplicate groups:")
for root, members in groups.items():
    if len(members) > 1:
        print(f"  Group [{root}]: {', '.join(members)} - merge into one ticket")
    else:
        print(f"  Unique: {members[0]}")

Expected output:

Candidate pairs after blocking: 3
  DUPLICATE (0.91): T-101 + T-102
    Both describe login/authentication page failing to load, same timeframe
  DUPLICATE (0.88): T-101 + T-104
    Both report login system down with loading failures
  DUPLICATE (0.85): T-103 + T-105
    Both report password reset emails not being delivered

Duplicate groups:
  Group [T-101]: T-101, T-102, T-104 - merge into one ticket
  Group [T-103]: T-103, T-105 - merge into one ticket
  Unique: T-104