Entity Resolution

Match policy records for the same person or household across insurance systems. Add up their coverage to spot totals above a risk limit, then choose which excess risk to transfer to another insurer within a budget.

---

Goals

UnderstandDetect

Reasoning types

GraphRules-basedPrescriptive

Experience level

Intermediate

Download ZIPView on GitHub

Browse files

• Directoryentity_resolution/

  • Directorydata/

    • ground_truth.csv
    • records.csv
  • Directorymodel/

    • __init__.py
    • schema.py
    • source.py
  • entity_resolution.py
  • pyproject.toml
  • README.md
  • runbook.md
  • template-docs.yaml

What this template is for

Insurers often store the same person or household as separate policyholder records, hiding combined coverage and accumulation risk. This template scores record matches, clusters high-confidence matches into resolved parties, and holds uncertain pairs for review. It then totals coverage, flags parties above a configurable limit, and selects reinsurance cessions within a premium budget. Adapt the matching fields, thresholds, exposure rules, and decision constraints for your workflow.

Explore the model

Explore the model

The model represents raw policy records, accepted and review-band matches, and the resolved insured parties used for accumulation control. Accepted matches create graph edges, while party membership carries each policy's coverage into the resolved exposure.

RecordCandidateMatchReviewPairResolvedPartyaccepts first recordaccepts second recordreviews first recordreviews second recordcontributes exposure to

RecordCandidateMatchReviewPairResolvedPartyaccepts first recordaccepts second recordreviews first recordreviews second recordcontributes exposure to

Record

A policyholder record and its coverage from one source system.

Properties

Name	Type
record_id (identity)	Integer
source_system	String
full_name	String
street	String
city	String
state	String
postal_code	String
created_at	String
coverage_amount	Float
date_of_birth	String
gov_id_last4	String
email	String
phone	String
entity_key	Integer

Relationships

accepts first record

CandidateMatch accepts a first Record

accepts second record

CandidateMatch accepts a second Record

reviews first record

ReviewPair holds a first Record for review

reviews second record

ReviewPair holds a second Record for review

contributes exposure to

Record contributes exposure to ResolvedParty

Preview source datarecords.csv · 51 rows; showing 4

Dirty policyholder records, matching identifiers, source systems, and per-policy coverage.

View data/records.csv

record_id	source_system	full_name	date_of_birth	gov_id_last4	email	phone	street	city	state	postal_code	created_at	coverage_amount
1005	HOME	Michael O'Sullivan	1972-06-25	7834	msullivan@workmail.com	(617) 555-0190	42 Beacon St	Boston	MA	02108	2019-05-22	480000
1006	AUTO	Mike O Sullivan	1972-06-25	7834	msullivan@workmail.com	6175550190	42 Beacon St Unit 5	Boston	MA	02108	2021-12-01	52000
1003	AUTO	Jennifer Walsh	1985-11-03	2261	jwalsh@outlook.com	(212) 555-0177	55 W 21st St	New York	NY	10010	2020-09-14	66000
1004	LIFE	Jen Walsh	1985-11-03	2261	jwalsh@outlook.com	212.555.0177	55 West 21st Street	New York	NY	10010	2022-11-30	740000

CandidateMatch

A pair of policy records with a match score high enough to link without manual review.

Properties

Name	Type
pair_id (identity)	Integer
score	Float
confidence_tier	String

Relationships

accepts first record

CandidateMatch accepts a first Record

accepts second record

CandidateMatch accepts a second Record

Preview source datarecords.csv · 51 rows; showing 4

Dirty policyholder records, matching identifiers, source systems, and per-policy coverage.

View data/records.csv

record_id	source_system	full_name	date_of_birth	gov_id_last4	email	phone	street	city	state	postal_code	created_at	coverage_amount
1005	HOME	Michael O'Sullivan	1972-06-25	7834	msullivan@workmail.com	(617) 555-0190	42 Beacon St	Boston	MA	02108	2019-05-22	480000
1006	AUTO	Mike O Sullivan	1972-06-25	7834	msullivan@workmail.com	6175550190	42 Beacon St Unit 5	Boston	MA	02108	2021-12-01	52000
1003	AUTO	Jennifer Walsh	1985-11-03	2261	jwalsh@outlook.com	(212) 555-0177	55 W 21st St	New York	NY	10010	2020-09-14	66000
1004	LIFE	Jen Walsh	1985-11-03	2261	jwalsh@outlook.com	212.555.0177	55 West 21st Street	New York	NY	10010	2022-11-30	740000

ReviewPair

A possible match between two policy records that the template sets aside for human review instead of linking automatically.

Properties

Name	Type
pair_id (identity)	Integer
score	Float

Relationships

reviews first record

ReviewPair holds a first Record for review

reviews second record

ReviewPair holds a second Record for review

Preview source datarecords.csv · 51 rows; showing 4

Dirty policyholder records, matching identifiers, source systems, and per-policy coverage.

View data/records.csv

record_id	source_system	full_name	date_of_birth	gov_id_last4	email	phone	street	city	state	postal_code	created_at	coverage_amount
1005	HOME	Michael O'Sullivan	1972-06-25	7834	msullivan@workmail.com	(617) 555-0190	42 Beacon St	Boston	MA	02108	2019-05-22	480000
1006	AUTO	Mike O Sullivan	1972-06-25	7834	msullivan@workmail.com	6175550190	42 Beacon St Unit 5	Boston	MA	02108	2021-12-01	52000
1003	AUTO	Jennifer Walsh	1985-11-03	2261	jwalsh@outlook.com	(212) 555-0177	55 W 21st St	New York	NY	10010	2020-09-14	66000
1004	LIFE	Jen Walsh	1985-11-03	2261	jwalsh@outlook.com	212.555.0177	55 West 21st Street	New York	NY	10010	2022-11-30	740000

ResolvedParty

One person or household as represented by one or more policy records. The template totals those records’ coverage to check the risk limit.

Properties

Name	Type
key (identity)	Integer
total_exposure	Float
excess	Float
premium	Float
cede	Float

Relationships

contributes exposure to

Record contributes exposure to ResolvedParty

CandidateMatch accepts a first Record

Connects an accepted match to the first of its two policy records.

Fields

Field	Value
Name	accepts first record
From	CandidateMatch
To	Record

CandidateMatch accepts a second Record

Connects an accepted match to the second of its two policy records.

Fields

Field	Value
Name	accepts second record
From	CandidateMatch
To	Record

ReviewPair holds a first Record for review

Connects a pair awaiting human review to the first of its two policy records.

Fields

Field	Value
Name	reviews first record
From	ReviewPair
To	Record

ReviewPair holds a second Record for review

Connects a pair awaiting human review to the second of its two policy records.

Fields

Field	Value
Name	reviews second record
From	ReviewPair
To	Record

Record contributes exposure to ResolvedParty

Connects each policy record to its resolved party, allowing the template to add up coverage across that party’s records.

Fields

Field	Value
Name	contributes exposure to
From	Record
To	ResolvedParty

records.csv

Dirty policyholder records, matching identifiers, source systems, and per-policy coverage.

Populates

Model item

CandidateMatch

Model item

Record

Model item

ReviewPair

Source

Path

data/records.csv

Kind

Local CSV

Preview: 51 rows; showing 4

record_id	source_system	full_name	date_of_birth	gov_id_last4	email	phone	street	city	state	postal_code	created_at	coverage_amount
1005	HOME	Michael O'Sullivan	1972-06-25	7834	msullivan@workmail.com	(617) 555-0190	42 Beacon St	Boston	MA	02108	2019-05-22	480000
1006	AUTO	Mike O Sullivan	1972-06-25	7834	msullivan@workmail.com	6175550190	42 Beacon St Unit 5	Boston	MA	02108	2021-12-01	52000
1003	AUTO	Jennifer Walsh	1985-11-03	2261	jwalsh@outlook.com	(212) 555-0177	55 W 21st St	New York	NY	10010	2020-09-14	66000
1004	LIFE	Jen Walsh	1985-11-03	2261	jwalsh@outlook.com	212.555.0177	55 West 21st Street	New York	NY	10010	2022-11-30	740000

Download and run the template

Before you start, install Python 3.10 or later and get access to a Snowflake account with the RAI Native App. Graph and Prescriptive reasoning are in Public Preview; ask your RelationalAI support representative to enable Prescriptive reasoning. Preview features are for evaluation and testing, not production applications. The template pins relationalai==1.13.0 in pyproject.toml and uses HiGHS without a separate solver license.

Use this sequence to run the bundled example:

1. Download the template

   Download the ZIP, unzip it, and enter the template directory:

   unzip entity_resolution.zip
   cd entity_resolution

2. Create a Python environment

   Create and activate a virtual environment, then update pip:

   python -m venv .venv
   source .venv/bin/activate
   python -m pip install --upgrade pip

3. Install the template

   Install the dependencies pinned in pyproject.toml:

   python -m pip install .

4. Configure your project

   Use the configuration builder in Start building with PyRel to create raiconfig.yaml in the template directory and verify your connection.

5. Run the template

   Run the bundled script from the template directory:

   python entity_resolution.py

   Output

   Auto-resolved 51 records into 31 insured parties.
   Households over the limit after RESOLUTION:       4
     -> ceded $927,000 of excess exposure for $111,240 premium (of $120,000)

   No single policy breaches the $1M limit, but resolution surfaces four over-limit households and the optimizer cedes the most excess it can afford within budget. See runbook.md for the full workflow.

See how it works

The Model Explorer shows the concepts for policy records, match pairs, and resolved parties, along with their relationships. The steps below show how the template loads records, scores possible matches, groups linked records, totals coverage, and chooses which excess risk to transfer. Use the excerpts to find the input mappings, match thresholds, risk limit, and budget you could adapt for your own data.

Hover over dotted-underlined code terms to highlight their source. You can also focus or tap each term.

1

Load the policy records

Each row of data/records.csv describes one policy, including identifying details and coverage amount. read_csv(...) reads the file, and Record.new(...) defines a separate Record instance from each row’s required fields (see Define base facts). In this loader, model.data(...).to_schema() omits a row if any field passed to Record.new(...) is empty, so including optional identifiers in that call could leave a whole policy out of the model. The optional-fields loop adds email, phone, and other identifiers separately, only for records that have those values.

model/source.py (lines 188-205)

def load_records() -> DataFrame:
    """Read policyholder records and load their base facts."""
    records = read_csv(
        DATA_DIR / "records.csv",
        dtype={"postal_code": str, "gov_id_last4": str},
    ).fillna("")
    records["coverage_amount"] = records["coverage_amount"].astype(float)
    model.define(Record.new(model.data(records[REQUIRED_COLS]).to_schema()))

    for column in OPTIONAL_COLS:
        present = records[records[column] != ""][["record_id", column]]
        present_data = model.data(present)
        model.define(
            getattr(Record.lookup(record_id=present_data.record_id), column)(
                getattr(present_data, column)
            )
        )
    return records

model/source.py (lines 188-205)

def load_records() -> DataFrame:
    """Read policyholder records and load their base facts."""
    records = read_csv(
        DATA_DIR / "records.csv",
        dtype={"postal_code": str, "gov_id_last4": str},
    ).fillna("")
    records["coverage_amount"] = records["coverage_amount"].astype(float)
    model.define(Record.new(model.data(records[REQUIRED_COLS]).to_schema()))

    for column in OPTIONAL_COLS:
        present = records[records[column] != ""][["record_id", column]]
        present_data = model.data(present)
        model.define(
            getattr(Record.lookup(record_id=present_data.record_id), column)(
                getattr(present_data, column)
            )
        )
    return records

2

Find and score possible matches

Comparing every policy with every other policy would waste work, so blocking_keys(...) groups records that share clues such as a phone number, date of birth, or surname and postal-code prefix. The candidate-pair loop collects each distinct pair once, and pair_score(...) scores the pairs using their identifying details. Pairs scoring at least AUTO_MERGE are accepted as matches, which the next stages can use to group policies under one insured party. REVIEW_FLOOR retains less certain pairs for human review instead of merging them, and pairs below that floor are omitted.

model/source.py (lines 208-235)

def build_candidate_pairs(records: DataFrame):
    """Block and score records, returning accepted and review-band pairs."""
    rows = {int(row["record_id"]): row for _, row in records.iterrows()}
    ids = sorted(rows)

    blocks: dict[str, list[int]] = {}
    for record_id in ids:
        for key in blocking_keys(rows[record_id]):
            blocks.setdefault(key, []).append(record_id)

    candidates: set[tuple[int, int]] = set()
    for members in blocks.values():
        for rec_a, rec_b in combinations(sorted(members), 2):
            candidates.add((rec_a, rec_b))

    accepted, review = [], []
    for pair_id, (rec_a, rec_b) in enumerate(sorted(candidates)):
        score = round(pair_score(rows[rec_a], rows[rec_b]), 4)
        pair = {
            "pair_id": pair_id,
            "rec_a": rec_a,
            "rec_b": rec_b,
            "score": score,
        }
        if score >= AUTO_MERGE:
            accepted.append(pair)
        elif score >= REVIEW_FLOOR:
            review.append(pair)

model/source.py (lines 208-235)

def build_candidate_pairs(records: DataFrame):
    """Block and score records, returning accepted and review-band pairs."""
    rows = {int(row["record_id"]): row for _, row in records.iterrows()}
    ids = sorted(rows)

    blocks: dict[str, list[int]] = {}
    for record_id in ids:
        for key in blocking_keys(rows[record_id]):
            blocks.setdefault(key, []).append(record_id)

    candidates: set[tuple[int, int]] = set()
    for members in blocks.values():
        for rec_a, rec_b in combinations(sorted(members), 2):
            candidates.add((rec_a, rec_b))

    accepted, review = [], []
    for pair_id, (rec_a, rec_b) in enumerate(sorted(candidates)):
        score = round(pair_score(rows[rec_a], rows[rec_b]), 4)
        pair = {
            "pair_id": pair_id,
            "rec_a": rec_a,
            "rec_b": rec_b,
            "score": score,
        }
        if score >= AUTO_MERGE:
            accepted.append(pair)
        elif score >= REVIEW_FLOOR:
            review.append(pair)

3

Keep accepted and review pairs separate

After scoring, one group contains accepted matches and the other contains uncertain pairs for review. CandidateMatch.new(...) uses Record.lookup(...) to define base facts connecting each accepted pair to its two existing policy records. ReviewPair.new(...) defines uncertain pairs separately, retaining their scores for a person to inspect.

model/source.py (lines 244-264)

def load_candidate_facts(matches: DataFrame, reviews: DataFrame) -> None:
    """Load accepted matches and review pairs into the model."""
    match_data = model.data(matches)
    model.define(
        CandidateMatch.new(
            pair_id=match_data.pair_id,
            rec_a=Record.lookup(record_id=match_data.rec_a),
            rec_b=Record.lookup(record_id=match_data.rec_b),
            score=match_data.score,
        )
    )

    review_data = model.data(reviews)
    model.define(
        ReviewPair.new(
            pair_id=review_data.pair_id,
            rec_a=Record.lookup(record_id=review_data.rec_a),
            rec_b=Record.lookup(record_id=review_data.rec_b),
            score=review_data.score,
        )
    )

model/source.py (lines 244-264)

def load_candidate_facts(matches: DataFrame, reviews: DataFrame) -> None:
    """Load accepted matches and review pairs into the model."""
    match_data = model.data(matches)
    model.define(
        CandidateMatch.new(
            pair_id=match_data.pair_id,
            rec_a=Record.lookup(record_id=match_data.rec_a),
            rec_b=Record.lookup(record_id=match_data.rec_b),
            score=match_data.score,
        )
    )

    review_data = model.data(reviews)
    model.define(
        ReviewPair.new(
            pair_id=review_data.pair_id,
            rec_a=Record.lookup(record_id=review_data.rec_a),
            rec_b=Record.lookup(record_id=review_data.rec_b),
            score=review_data.score,
        )
    )

4

Group linked records into insured parties

To create a graph, Graph(...) treats each policy record as a node, while CandidateMatch supplies an edge for each accepted pair. This graph is undirected because either record in a match can lead to the other. graph.weakly_connected_component() groups records connected by accepted matches, including chains where the first and last records never matched directly. Record.entity_key(...) assigns each record its group’s representative ID, giving later calculations a key for the insured party.

entity_resolution.py (lines 76-91)

def cluster_accepted_matches() -> tuple[int, int]:
    """Cluster records connected by accepted match edges."""
    graph = Graph(model, directed=False, weighted=False, node_concept=Record)

    match_ref = CandidateMatch.ref()
    rec_a, rec_b = Record.ref(), Record.ref()
    model.where(match_ref.rec_a == rec_a, match_ref.rec_b == rec_b).define(
        graph.Edge.new(src=rec_a, dst=rec_b)
    )

    graph.Node.entity_id = graph.weakly_connected_component()

    rep_node = Record.ref()
    model.where(Record.entity_id == rep_node).define(
        Record.entity_key(rep_node.record_id)
    )

entity_resolution.py (lines 76-91)

def cluster_accepted_matches() -> tuple[int, int]:
    """Cluster records connected by accepted match edges."""
    graph = Graph(model, directed=False, weighted=False, node_concept=Record)

    match_ref = CandidateMatch.ref()
    rec_a, rec_b = Record.ref(), Record.ref()
    model.where(match_ref.rec_a == rec_a, match_ref.rec_b == rec_b).define(
        graph.Edge.new(src=rec_a, dst=rec_b)
    )

    graph.Node.entity_id = graph.weakly_connected_component()

    rep_node = Record.ref()
    model.where(Record.entity_id == rep_node).define(
        Record.entity_key(rep_node.record_id)
    )

5

Find parties above the coverage limit

Records with the same graph-derived party key may hold separate policies for one insured person or household. ResolvedParty.new(...) creates a party for each key, and record.resolved_party(...) connects its policy records. To aggregate and group data, ResolvedParty.total_exposure(...) adds the coverage amounts of those records for each party. A comparison with ACCUMULATION_LIMIT flags parties above the risk limit, and ResolvedParty.excess(...) measures how much coverage exceeds it for the reinsurance decision.

entity_resolution.py (lines 110-129)

    model.define(ResolvedParty.new(key=Record.entity_key))
    record, party = Record.ref(), ResolvedParty.ref()
    model.where(record.entity_key == party.key).define(
        record.resolved_party(party)
    )

    party_record = Record.ref()
    model.where(party_record.entity_key == ResolvedParty.key).define(
        ResolvedParty.total_exposure(
            aggregates.sum(party_record.coverage_amount).per(ResolvedParty)
        )
    )
    model.define(ResolvedParty.is_over_limit(ResolvedParty)).where(
        ResolvedParty.total_exposure > ACCUMULATION_LIMIT
    )
    model.where(ResolvedParty.total_exposure > ACCUMULATION_LIMIT).define(
        ResolvedParty.excess(
            ResolvedParty.total_exposure - ACCUMULATION_LIMIT
        )
    )

entity_resolution.py (lines 110-129)

    model.define(ResolvedParty.new(key=Record.entity_key))
    record, party = Record.ref(), ResolvedParty.ref()
    model.where(record.entity_key == party.key).define(
        record.resolved_party(party)
    )

    party_record = Record.ref()
    model.where(party_record.entity_key == ResolvedParty.key).define(
        ResolvedParty.total_exposure(
            aggregates.sum(party_record.coverage_amount).per(ResolvedParty)
        )
    )
    model.define(ResolvedParty.is_over_limit(ResolvedParty)).where(
        ResolvedParty.total_exposure > ACCUMULATION_LIMIT
    )
    model.where(ResolvedParty.total_exposure > ACCUMULATION_LIMIT).define(
        ResolvedParty.excess(
            ResolvedParty.total_exposure - ACCUMULATION_LIMIT
        )
    )

6

Choose which excess risk to transfer

An insurer can pay a reinsurer to take on the risk above each party’s coverage limit, but it must stay within a budget. Problem(model, Float) sets up the decision, and problem.solve_for(...) gives each over-limit party a yes-or-no choice to transfer its excess. problem.satisfy(...) adds a constraint that caps the total premium paid to the reinsurer at REINSURANCE_BUDGET. problem.maximize(...) aims to transfer the most excess risk, and problem.solve(...) runs the solver, though a small budget may leave some excess with the insurer.

entity_resolution.py (lines 135-159)

def choose_reinsurance_cessions():
    """Solve the reinsurance-budget knapsack for breached parties."""
    problem = Problem(model, Float)
    problem.solve_for(
        ResolvedParty.cede,
        where=[ResolvedParty.is_over_limit()],
        name=["cede", ResolvedParty.key],
        type="bin",
        lower=0.0,
        upper=1.0,
    )
    problem.satisfy(
        model.require(
            aggregates.sum(
                ResolvedParty.premium * ResolvedParty.cede
            )
            <= REINSURANCE_BUDGET
        ),
        name=["reinsurance_budget"],
    )
    problem.maximize(
        aggregates.sum(ResolvedParty.excess * ResolvedParty.cede)
    )
    problem.solve("highs", time_limit_sec=60)
    return problem.solve_info()

entity_resolution.py (lines 135-159)

def choose_reinsurance_cessions():
    """Solve the reinsurance-budget knapsack for breached parties."""
    problem = Problem(model, Float)
    problem.solve_for(
        ResolvedParty.cede,
        where=[ResolvedParty.is_over_limit()],
        name=["cede", ResolvedParty.key],
        type="bin",
        lower=0.0,
        upper=1.0,
    )
    problem.satisfy(
        model.require(
            aggregates.sum(
                ResolvedParty.premium * ResolvedParty.cede
            )
            <= REINSURANCE_BUDGET
        ),
        name=["reinsurance_budget"],
    )
    problem.maximize(
        aggregates.sum(ResolvedParty.excess * ResolvedParty.cede)
    )
    problem.solve("highs", time_limit_sec=60)
    return problem.solve_info()