Introduction
As your agents begin to read and write across apps, warehouses, and streams, identity fragmentation becomes the silent killer: the same customer appears as five records; suppliers are duplicated across ERPs; employees show up as both contractors and full-timers. This article delivers a Master Data & Entity Resolution Agent that deduplicates and links records into golden entities—safely, audibly, and with receipts—so every downstream agent (support, billing, analytics) acts on a consistent truth.
The Use Case
Traditional MDM projects are slow and centralized. Meanwhile, agents need good-enough, explainable resolution now. Our agent ingests candidate records (customers/suppliers/products), computes match proposals (deterministic + fuzzy), requests approvals for risky merges, executes merges through governed APIs, and emits linkage edges for analytics. Crucially, it never claims success without receipts (merge job IDs, link IDs, ticket IDs) and it records why two records were linked (feature attributions and thresholds).
Prompt Contract (agent interface)
# file: contracts/entity_resolution_v1.yaml
role: "EntityResolutionAgent"
scope: >
Propose, approve, and execute entity links/merges across systems using governed rules.
Ask once if critical fields are missing (domain, records[], ruleset, risk_tolerance).
Never assert success without a receipt (link_id, merge_id, ticket id).
inputs:
domain: enum["customer","supplier","employee","product"]
ruleset: string # e.g., "cust_rules_v3"
risk_tolerance: enum["low","medium","high"]
records: # candidates from one or more systems
- {system: string, id: string, attrs: object}
output:
type: object
required: [summary, proposals, decisions, citations, next_steps, tool_proposals]
properties:
summary: {type: string, maxWords: 100}
proposals:
type: array
items:
type: object
required: [pair, score, features, action]
properties:
pair: {type: array, items: string} # ["crm:123","billing:987"]
score: {type: number} # 0..1
features: {type: array, items: string} # e.g., "email_exact","name_jaro=0.92"
action: {type: string, enum: ["auto_link","review","reject"]}
decisions:
type: array
items:
type: object
required: [pair, decision, reason, receipt]
properties:
pair: {type: array, items: string}
decision: {type: string, enum: ["linked","merged","rejected","needs_review"]}
reason: {type: string}
receipt: {type: string}
citations: {type: array, items: string} # ruleset id, policy claims
next_steps: {type: array, items: string, maxItems: 6}
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name:
type: string
enum: [LoadRules, ComputeFeatures, ProposeMatches, RequestApproval,
CreateLink, MergeRecords, OpenTicket, EmitLineage]
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "mdm_policy.v4"
citation_rule: "Minimal-span references to ruleset id and feature thresholds."
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from typing import List, Dict, Optional
class LoadRulesArgs(BaseModel):
ruleset: str
class ComputeFeaturesArgs(BaseModel):
domain: str
records: List[Dict] # [{system,id,attrs}]
class ProposeMatchesArgs(BaseModel):
features: List[Dict] # output from ComputeFeatures
risk_tolerance: str # "low"|"medium"|"high"
class RequestApprovalArgs(BaseModel):
pair: List[str]
score: float
features: List[str]
approvers: List[str]
class CreateLinkArgs(BaseModel):
canonical_id: str
member_id: str
domain: str
class MergeRecordsArgs(BaseModel):
domain: str
into_id: str
from_id: str
field_strategy: Dict[str,str] # {"email":"prefer_non_null","name":"longest","address":"most_recent"}
class OpenTicketArgs(BaseModel):
title: str
description: str
severity: str
owners: List[str]
class EmitLineageArgs(BaseModel):
domain: str
canonical_id: str
members: List[str]
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str # e.g., link_id, merge_id, ticket id
message: str = ""
data: Optional[Dict] = None
# adapters.py (demo logic; wire to your MDM/CRM/ERP/catalog in prod)
from tools import *
import uuid, random
RULES = {
"cust_rules_v3": {
"auto_link_threshold": 0.93,
"review_threshold": 0.80,
"feature_weights": {"email_exact":0.6,"phone_e164":0.25,"name_jaro":0.1,"addr_city_exact":0.05}
}
}
APPROVERS = {"customer": ["dataops@company","crm-owner@company"]}
def load_rules(a: LoadRulesArgs) -> ToolReceipt:
return ToolReceipt(tool="LoadRules", ok=True, ref=a.ruleset, data=RULES.get(a.ruleset, {}))
def compute_features(a: ComputeFeaturesArgs) -> ToolReceipt:
feats = []
recs = a.records
# toy pairwise features for first two records
for i in range(len(recs)):
for j in range(i+1, len(recs)):
r1, r2 = recs[i], recs[j]
email_exact = int(r1["attrs"].get("email","").lower()==r2["attrs"].get("email","").lower() and r1["attrs"].get("email"))
phone_e164 = int(r1["attrs"].get("phone")==r2["attrs"].get("phone") and r1["attrs"].get("phone"))
name_jaro = round(random.uniform(0.7, 0.99), 2) # stand-in
feats.append({"pair":[f"{r1['system']}:{r1['id']}", f"{r2['system']}:{r2['id']}"],
"features":[f"email_exact={email_exact}", f"phone_e164={phone_e164}", f"name_jaro={name_jaro}"]})
return ToolReceipt(tool="ComputeFeatures", ok=True, ref="feats-1", data={"features":feats})
def propose_matches(a: ProposeMatchesArgs) -> ToolReceipt:
# toy scorer reading features
out = []
for f in a.features:
score = 0.0
for feat in f["features"]:
k,v = feat.split("="); v = float(v)
if k=="email_exact": score += 0.6*v
if k=="phone_e164": score += 0.25*v
if k=="name_jaro": score += 0.15*v
action = "reject"
if score >= 0.93: action = "auto_link"
elif score >= 0.80: action = "review" if a.risk_tolerance!="high" else "auto_link"
out.append({"pair":f["pair"], "score":round(score,2), "features":f["features"], "action": action})
return ToolReceipt(tool="ProposeMatches", ok=True, ref="prop-1", data={"proposals":out})
def request_approval(a: RequestApprovalArgs) -> ToolReceipt:
return ToolReceipt(tool="RequestApproval", ok=True, ref="APR-001", message="Approval requested", data={"approvers":a.approvers})
def create_link(a: CreateLinkArgs) -> ToolReceipt:
return ToolReceipt(tool="CreateLink", ok=True, ref=f"LINK-{uuid.uuid4().hex[:8]}", message="Linked to canonical")
def merge_records(a: MergeRecordsArgs) -> ToolReceipt:
return ToolReceipt(tool="MergeRecords", ok=True, ref=f"MERGE-{uuid.uuid4().hex[:8]}", message="Merged into canonical")
def open_ticket(a: OpenTicketArgs) -> ToolReceipt:
return ToolReceipt(tool="OpenTicket", ok=True, ref="MDM-217", message="Review ticket opened")
def emit_lineage(a: EmitLineageArgs) -> ToolReceipt:
return ToolReceipt(tool="EmitLineage", ok=True, ref=f"EDGE-{uuid.uuid4().hex[:8]}", message="Lineage emitted")
Join the conversation! Your thoughts help the community grow.