Metadata-Version: 2.5
Name: hsds-entity-resolution
Version: 2.5.0
Summary: HSDS entity resolution: candidate generation, scoring, a Pair Judge contract and clustering, with a Dagster component.
Project-URL: Homepage, https://github.com/211-Connect/hsds-entity-resolution
Project-URL: Repository, https://github.com/211-Connect/hsds-entity-resolution
Project-URL: Issues, https://github.com/211-Connect/hsds-entity-resolution/issues
Keywords: dagster,deduplication,entity-resolution,hsds
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.10
Requires-Dist: dagster<1.13,>=1.11
Requires-Dist: numpy<3,>=1.26
Requires-Dist: polars<2,>=1.18
Requires-Dist: pydantic<3,>=2
Requires-Dist: tldextract>=5.3.1
Description-Content-Type: text/markdown

# hsds-entity-resolution

Reusable [Dagster](https://dagster.io) component for incremental entity resolution on
[HSDS](https://docs.openreferral.org/en/latest/) (Human Services Data Specification) records.

**Install:**

```bash
pip install hsds-entity-resolution
```

---

## What this package does

Community 211 directories and social-services platforms often ingest the same organizations
and services from multiple data partners. Over time those records diverge — different spellings,
phone formats, partial addresses — making it hard to know which rows describe the same real-world
entity.

`hsds-entity-resolution` provides a single Dagster component, `EntityResolutionComponent`, that
runs an incremental seven-stage entity-resolution pipeline on your HSDS data:

| Stage | What happens |
|---|---|
| **Clean entities** | Normalize contact fields, compute content hashes, detect adds/changes/removals since the last run |
| **Generate candidates** | Admit a pair when it shares an Informative Key (name, phone, email, website, address) or its embedding similarity reaches one floor, unless a caller's Structural Exclusion vetoes it |
| **Score candidates** | Weighted combination of deterministic overlap signals and NLP fuzzy name/description matching |
| **Judge pairs** *(optional)* | A caller-supplied `PairJudge` answers Same Site, Same Offering, physical delivery and Organization Relation as probabilities for every scored pair |
| **Apply mitigation** | Carry forward stable pairs, retire pairs for removed entities, detect pair identity continuity |
| **Cluster pairs** | Greedy correlation clustering groups high-confidence duplicate pairs into clusters |
| **Materialize review queue** | Pairs that score above the *maybe* threshold but below *duplicate* are surfaced for human review |
| **Prepare artifacts** | Package all stage outputs into a typed bundle ready for downstream persistence |

The component is **storage-agnostic**: it accepts and emits Polars DataFrames, so you wire it
into whatever persistence layer your Dagster project uses.

---

## Prerequisites

- Python 3.10–3.13
- A [Dagster](https://dagster.io) project managed with the
  [`dg` CLI](https://docs.dagster.io/guides/build/projects-and-components/components/)
- Four upstream Dagster assets that supply denormalized HSDS entity data (see [Input assets](#input-assets))

---

## Quickstart

### 1. Discover the component

```bash
dg list components --package hsds_entity_resolution
```

### 2. Add to your component YAML

```yaml
type: hsds_entity_resolution.dagster.components.EntityResolutionComponent
attributes:
  team_id: my_team
  scope_id: regional
  entity_type: organization
```

That's enough to wire the component into your Dagster asset graph. The four required upstream
assets default to the keys `organization_entities`, `service_entities`,
`previous_entity_index`, and `previous_pair_state_index` — override them with the
`*_asset_key` attributes if your project uses different names.

---

## Input assets

The component depends on four upstream assets. Each must yield a Polars `DataFrame`
(or be coercible to one).

### `organization_entities` and `service_entities`

Denormalized HSDS entity records. Required columns:

| Column | Type | Description |
|---|---|---|
| `entity_id` | `str` | Stable unique identifier for this record |
| `source_schema` | `str` | Tenant or source identifier (e.g. `region_a`) |
| `name` | `str` | Entity name used for NLP scoring |
| `description` | `str` | Entity description used for NLP scoring |
| `emails` | `list[str]` | Normalized email addresses |
| `phones` | `list[str]` | Normalized phone numbers |
| `websites` | `list[str]` | Normalized domain names |
| `locations` | `list[dict]` | Structured address/location objects |
| `taxonomies` | `list[dict]` | AIRS taxonomy code objects |
| `identifiers` | `list[dict]` | External identifier objects |
| `embedding_vector` | `list[float]` | Pre-computed text embedding (e.g. 384-dim) |

Optional passthrough columns (`display_name`, `display_description`, `alternate_name`,
`short_description`, `application_process`, `fees_description`, `eligibility_description`,
`resource_writer_name`, `assured_date`, `assurer_email`, `original_id`,
`organization_original_id`, `organization_name`, `organization_id`,
`services_rollup`) are forwarded unchanged to the normalized entity cache and
persistence artifacts.

Optional judge evidence columns (2.4.0) are passed through, never scored or hashed, and
read only by the Pair Judge state:

| Column | Type | Description |
|---|---|---|
| `service_areas` | `list[{"level": str, "place": str}]` | Structured extent: `level` is `national`, `state`, `county`, `city` or `postal_code` (`zip`, `zip_code` accepted); `place` names it, e.g. `"Cook County, IL"`, `"60601"`. Derive both from structured extent, not from the area's free-text name |
| `schedules` | `list[{"description": str, "scope": "service" \| "location"}]` | One canonical English text per schedule; `scope` says whether it hangs on the service or on a service-at-location |
| `parent_organization_id` | `str` | HSDS `organization.parent_organization_id`, on organization rows |

### `previous_entity_index`

The `normalized_organization` or `normalized_service` output from the *previous* run of this
component. Pass an empty `pl.DataFrame()` on first run.

### `previous_pair_state_index`

The `pair_state_index` output from the *previous* run of this component. Pass an empty
`pl.DataFrame()` on first run. The component uses this to carry forward stable pair scores
and detect pair identity continuity across incremental runs.

---

## Component attributes

| Attribute | Type | Default | Description |
|---|---|---|---|
| `team_id` | `str` | `"hsds"` | Team identifier written into run metadata |
| `scope_id` | `str` | `"default"` | Deployment or region identifier |
| `entity_type` | `"organization" \| "service"` | `"organization"` | Which entity type this instance processes |
| `policy_version` | `str` | `"hsds-er-v1"` | Scoring policy version tag |
| `model_version` | `str` | `"embedding-only-v1"` | Embedding model version tag |
| `explicit_backfill` | `bool` | `False` | Force a full re-run even when no entity changes are detected |
| `organization_entities_asset_key` | `str` | `"organization_entities"` | Asset key for upstream org entities |
| `service_entities_asset_key` | `str` | `"service_entities"` | Asset key for upstream service entities |
| `previous_entity_index_asset_key` | `str` | `"previous_entity_index"` | Asset key for previous-run entity cache |
| `previous_pair_state_index_asset_key` | `str` | `"previous_pair_state_index"` | Asset key for previous-run pair state |
| `output_asset_prefix` | `list[str]` | `[]` | Prefix prepended to all output asset keys |
| `constants_overrides` | `dict` | `{}` | Deep-merge override for scoring constants (see [Tuning](#tuning-scoring-constants)) |

---

## Output assets

The component emits fifteen assets. Using `output_asset_prefix: ["my_prefix"]` namespaces
all keys under `my_prefix/`.

| Asset key | Description |
|---|---|
| `normalized_organization` | Cleaned, normalized organization records with content hashes |
| `normalized_service` | Cleaned, normalized service records with content hashes |
| `entity_delta_summary` | Single-row summary of added / changed / removed entity counts |
| `removed_entity_ids` | Entity IDs that were present last run but are absent this run |
| `candidate_pairs` | Raw blocked candidate pairs before scoring |
| `scored_pairs` | Candidate pairs with composite scores and tier labels (`duplicate`, `maybe`, `non_duplicate`) |
| `pair_reasons` | Per-feature evidence rows for every scored pair |
| `mitigation_events` | Pairs reassigned due to entity changes between runs |
| `removed_pair_ids` | Pair IDs retired this run (entity removed or score dropped) |
| `pair_id_remap` | Maps old pair IDs to new ones when pair identity shifts |
| `clusters` | Entity groups identified as duplicates via correlation clustering |
| `cluster_pairs` | Edges within each cluster |
| `pair_state_index` | Full pair state — feed this back as `previous_pair_state_index` next run |
| `review_queue_items` | Borderline pairs above the *maybe* threshold, ready for human review |
| `run_summary` | Single-row run metrics (candidate count, duplicate count, cluster count, etc.) |
| `persistence_artifact_bundle` | Dict containing all stage outputs packaged for bulk persistence |

---

## Tuning scoring constants

The default thresholds are calibrated for HSDS data. Override any constant via
`constants_overrides` without subclassing:

```yaml
type: hsds_entity_resolution.dagster.components.EntityResolutionComponent
attributes:
  team_id: my_team
  scope_id: regional
  entity_type: organization
  constants_overrides:
    scoring:
      duplicate_threshold: 0.85
      maybe_threshold: 0.70
      nlp:
        fuzzy_threshold: 0.90
```

Key thresholds:

| Constant | Default (org) | Default (service) | Description |
|---|---|---|---|
| `scoring.duplicate_threshold` | `0.82` | `0.70` | Minimum score to auto-cluster as duplicate |
| `scoring.maybe_threshold` | `0.68` | `0.62` | Minimum score to send to review queue |
| `scoring.deterministic_section_weight` | `0.5625` | `0.50` | Weight of overlap-signal section |
| `scoring.nlp_section_weight` | `0.4375` | `0.50` | Weight of NLP fuzzy-match section; the two must sum to 1 |
| `scoring.nlp.fuzzy_threshold` | `0.88` | `0.86` | Minimum name similarity to count as NLP match |
| `blocking.similarity_threshold` | `0.75` | `0.75` | The embedding floor: a pair at or above it is a candidate on similarity alone |
| `blocking.max_candidates_per_entity` | `50` | `125` | Maximum embedding-floor candidates per anchor (shared-key candidates are not capped by it) |
| `blocking.exemption_floor` | unset | unset | Lower edge of the band below the floor where a `floor_exemption` reason admits a pair (2.5.0) |

---

## Multiple instances

Deploy one component instance per entity type and scope:

```yaml
# organizations
type: hsds_entity_resolution.dagster.components.EntityResolutionComponent
attributes:
  team_id: my_team
  scope_id: regional
  entity_type: organization
  output_asset_prefix: ["org_er"]

---

# services
type: hsds_entity_resolution.dagster.components.EntityResolutionComponent
attributes:
  team_id: my_team
  scope_id: regional
  entity_type: service
  output_asset_prefix: ["svc_er"]
```

---

## Pair Judge contract

The engine fixes what a judge reads and what it answers; it ships no questions or model.

- `hsds_entity_resolution.judge.PairState`: both records as HSDS text (name, alternate
  name, description, short description, eligibility, fees, application process,
  taxonomy names, sites, phones as digits, websites, organization name and
  description, and from 2.4.0 emails, organization identifiers, assurance date and
  assurer, service areas and schedules), a code-computed site comparison word
  (`same address`, `same city`, `different`, `unknown`), the prior score as a bucket
  in words, and one free-text `source_profile` slot per side. Two more facts are
  computed in code and sent as sentences, each left out when there is nothing to say:
  `assurance_recency` names the more recently assured record, so the other may be
  outdated, and says outright that assurance dates are not evidence of whether the
  records match; `organization_link` names a shared organization record (services
  with the same `organization_id`) or a parent (organizations, via
  `parent_organization_id`).
- Record evidence (2.4.0). `service_areas` lists `{level, places, omitted}` coarsest
  first, at most `MAX_PLACES_PER_LEVEL` (25) places per level; a `national` row naming
  only the country ("USA") is dropped when the record lists a narrower area, and an
  empty list means the area is unknown. `schedules` lists `{scope, text,
  low_information}`; text that names no hours ("Same as site hours", "Varies", "Call
  for hours") is marked `low_information`. `identifiers` lists `{type, identifier}`. The engine fills the HSDS fields and leaves the
  profile slots empty unless you pass `source_profiles={source_schema: text}`.
- `hsds_entity_resolution.judge.PairAnswers`: `same_site`, `same_offering`,
  `physically_delivered` (probabilities; the last two are null for organization
  pairs), and `organization_relation` as a distribution over `same`,
  `parent_and_chapter`, `affiliated`, `unrelated`, `cannot_tell`, plus `model_id` and
  `question_set_version`. The judge never answers "duplicate"; compose that in your
  own code. Any answer may be null when the judge did not ask that question (a judge
  that asks one question at a time and stops once your policy is decided); at least
  one is answered.
- `hsds_entity_resolution.judge.PairJudge`: a protocol with `judge(states) -> answers`
  and a `state_token_budget`. Size states for the smallest model you may switch to
  (`SMALLEST_MODEL_STATE_BUDGET_TOKENS`, 32k).
- `ReferenceJudge` answers 0.5 everywhere and a uniform relation, so the pipeline runs
  end to end without a model.

```python
from hsds_entity_resolution.core import run_incremental
from hsds_entity_resolution.judge import ReferenceJudge

result = run_incremental(..., judge=ReferenceJudge(), source_profiles={"source_a": "..."})
result.judge_answers  # one row per scored pair, JUDGE_ANSWERS_SCHEMA
```

`judge_scored_pairs(...)` runs the same stage on its own, for hosts that run the judge
as a separate step.

## Candidate admission

One generic rule decides which pairs are scored (2.1.0):

- **Informative Keys.** Two records that share a value of `name`, `phone`, `email`,
  `website` or `address` (street line required) are a candidate, whatever their
  embedding similarity, provided the field is informative in both records' source
  schemas. A field is informative where its values mostly identify a record; a value
  repeated across hundreds of rows (a category label used as a name) is not.
- **Embedding floor.** Records at or above `blocking.similarity_threshold` are a
  candidate, up to `blocking.max_candidates_per_entity` per anchor.
- **Floor exemption (2.5.0).** With `blocking.exemption_floor` set (at most
  `blocking.similarity_threshold`) and a caller `floor_exemption(entity_a, entity_b) ->
  reason | None`, a pair whose similarity is in `[exemption_floor, similarity_threshold)`
  is a candidate when the callable names a reason (for example two services of one
  organization). It carries the reason code `floor_exempt:<reason>` and blocking rule
  `floor_exempt`, is still subject to the Structural Exclusion, and counts against
  `max_candidates_per_entity` after every pair at or above the floor. Pairs only the
  exemption admitted are counted by reason in `floor_exempt_counts`.
- **Structural Exclusions.** A caller-supplied callable can veto any pair. Excluded
  pairs are returned as `excluded_pairs` with the reason, never dropped silently.

Pass both to `run_incremental` (or `run_incremental_until_candidates`):

```python
from hsds_entity_resolution.core.admission import build_informative_key_table

informative_keys = build_informative_key_table(
    ratios={"SOURCE_A": {"name": 0.12, "phone": 0.74}},  # distinct-value ratios you measure
    cutoff=0.3,
    overrides={"SOURCE_B": {"name": True}},
)

def exclude(entity_a, entity_b):
    return "different_by_construction" if ... else None

result = run_incremental(..., informative_keys=informative_keys, structural_exclusion=exclude)
result.excluded_pairs  # pair_key, entity ids, schemas, similarity, exclusion_reason
```

Without them every key field is informative and nothing is excluded. Scoring does not
depend on how a pair was admitted: `pair_rules` still pick scoring overrides from the
entity type and the two source schemas. `source_policy.admission_rules` and
`blocking.overlap_prefilter_channels` were removed; passing them fails validation with a
migration message.

## Normalisers

`hsds_entity_resolution.normalize` holds pure, idempotent functions on plain strings —
`normalize_phone` (digits plus extension), `normalize_email`, `normalize_url`,
`normalize_address_component`, `normalize_postal_code` — usable outside the engine.

## Migrating from `hsds-record-matcher` 1.x

- **Install name.** The distribution is now `hsds-entity-resolution`, matching the
  import name `hsds_entity_resolution`. Uninstall `hsds-record-matcher` first.
- **ML scoring section removed (1.2.0).** `scoring.ml`, `scoring.calibration`,
  `scoring.ml_section_weight` and pair-rule `ml_section_weight` / `ml_gate_threshold`
  now fail validation. `deterministic_section_weight + nlp_section_weight` must equal
  1.0; to keep your scores, divide each by their old sum. Scored pairs no longer carry
  `ml_section_score`, `legacy_confidence_score`, `shadow_confidence_score`,
  `shadow_log_odds` or `calibration_version`, and `run_incremental` no longer takes
  `taxonomy_embeddings`.
- **Judge stage (2.0.0).** `run_incremental` accepts `judge` and `source_profiles`;
  without a judge every output is unchanged.
- **Unasked answers (2.3.0).** Every `PairAnswers` answer, `same_site` and
  `organization_relation` included, may be `None`, and `organization_relation_choice`
  returns `None` when the relation was not asked. Service pairs no longer require
  `same_offering` and `physically_delivered`. Code that reads the answers must handle
  `None`.
- **Judge evidence (2.4.0).** `RecordState` gains `emails`, `identifiers`,
  `assured_date`, `assurer_email`, `service_areas` and `schedules`; `PairState` gains
  `assurance_recency` and `organization_link`. All default to empty, and the input
  columns behind them are optional, so existing callers keep working. A state's
  serialised form gains the new keys, so anything that fingerprints the state changes
  once for every record.
- **Floor exemption (2.5.0).** `run_incremental`, `run_incremental_until_candidates`
  and `generate_candidates` accept `floor_exemption`, and `BlockingConfig` gains
  `exemption_floor`. Both default to off; without both, admission is byte-identical to
  2.4.0. `GenerateCandidatesResult` and `IncrementalRunResult` gain
  `floor_exempt_counts` (reason → pairs only the exemption admitted), empty by default.
  `BlockingConfig.model_dump()` gains the `exemption_floor` key (`None`), so anything
  that fingerprints a dumped config changes once.
- **Logging.** Core modules log through `logging.getLogger("hsds_entity_resolution...")`
  instead of Dagster's logger. Add `hsds_entity_resolution` to your Dagster
  `python_logs.managed_python_loggers` to see them in the Dagster UI.

## Links

- **Source:** [github.com/211-Connect/hsds-entity-resolution](https://github.com/211-Connect/hsds-entity-resolution)
- **Issues:** [github.com/211-Connect/hsds-entity-resolution/issues](https://github.com/211-Connect/hsds-entity-resolution/issues)
- **PyPI:** [pypi.org/project/hsds-entity-resolution](https://pypi.org/project/hsds-entity-resolution/)
