# Run this notebook from a clone of the ddharmon repo: its bundled example data lives in
# data/examples/, and every path below is relative to the repo root. This installs ddharmon
# and the extras the pipeline needs (embeddings → sentence-transformers, llm → anthropic Batch
# API, clustering, bertopic) directly from this repo — no external or dev references.
%pip install -q ".[embeddings,llm,clustering,bertopic]"

import json
import logging
from pathlib import Path

from ddharmon.ingestion import load_dictionary
from ddharmon.embedding import SentenceTransformerProvider, embed_dictionary
from ddharmon.harmonization import (
    harmonize_dictionaries,
    assemble_verdicts,
    write_prompts_jsonl,
    write_buckets,
    export_eitl_queue,
)
from ddharmon.llm import submit_and_wait  # Anthropic Batch API (needs ANTHROPIC_API_KEY)

try:
    from ddharmon.ingestion import preprocess_dictionary
except ImportError:
    preprocess_dictionary = None

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("huggingface_hub").setLevel(logging.WARNING)
print("Imports OK")

Imports OK

DATA_DIR = Path("data/examples")  # bundled, publicly-available example dictionaries

# CDEs participate in clustering as a cohort. Ships with the full NIH CDE repo (~22.7k CDEs);
# rebuild or subset it with: python scripts/flatten_cde_repo.py <input.json> <output.tsv>
CDE_FILE = DATA_DIR / "all_cdes_flat.tsv"

# Two public example cohorts ship with the repo. Replace these with your own dictionaries —
# each loader just maps your columns onto load_dictionary's named kwargs.
loaders = {
    "NIH_CDE": (CDE_FILE, dict(
        variable_name="designation", field_id="tinyId",
        description="definition", question_text="question_text",
        data_type="datatype", value_encoding="permissible_values",
        category="classification", standard_code="concept_codes",
        embed_variable_name=True,
    )),
    "AllOfUs": (DATA_DIR / "all_of_us_surveys.csv", dict(
        variable_name="Item Concept", description="Field Label",
        category="Survey", data_type="Field Type",
        value_encoding="Choices, Calculations, OR Slider Labels",
        question_text="Field Label",
    )),
    "CLSA": (DATA_DIR / "clsa_baseline.csv", dict(
        variable_name="name", short_label="label:en", question_text="question:en",
        description="comment:en", category="table", data_type="valueType",
        units="unit", value_encoding="value_encoding",
    )),
    # Bring your own: copy the dict above, point it at data/examples/<your_file>.csv,
    # and map your column names to these kwargs.
}

cohorts = {}
for name, (path, kwargs) in loaders.items():
    if not path.exists():
        print(f"{name}: SKIPPED (not found: {path})")
        continue
    dd = load_dictionary(path, cohort_name=name, **kwargs)
    if preprocess_dictionary is not None:
        dd = preprocess_dictionary(dd)
    cohorts[name] = dd
    print(f"{name}: {dd.field_count} fields")

print(f"\nTotal: {sum(d.field_count for d in cohorts.values())} fields across {len(cohorts)} cohorts")

INFO: Column role mapping: {'designation': 'variable_name', 'tinyId': 'field_id', 'definition': 'description', 'datatype': 'data_type', 'classification': 'category', 'permissible_values': 'value_encoding', 'concept_codes': 'standard_code', 'question_text': 'question_text'}

INFO: Parsed 22743 fields from 22743 CSV rows

INFO: Hierarchy detection: 0 groups found, 0 synthetic parents created

INFO: load_dictionary(all_cdes_flat.tsv): 22269 fields in 1.76s

INFO: Unicode normalization: fixed 70 fields

INFO: Placeholder description replaced: "Complete the question for all hospitalized pulmonary embolism (PE)/deep vein thr" (11 fields)

INFO: Placeholder description replaced: "From CDC COVID-19 Community survey question bank (DRAFT)" (34 fields)

INFO: Placeholder description replaced: "NLM disclaimer: Consistent with Executive Order 14168, "gender" should not be us" (15 fields)

INFO: Placeholder description replaced: "No definition" (19 fields)

INFO: Placeholder description replaced: "Question is reversed scored" (17 fields)

provider = SentenceTransformerProvider()
embedded = []
for name, dd in cohorts.items():
    ed = embed_dictionary(dd, provider=provider)
    embedded.append(ed)
    print(f"  {name}: {len(ed.embeddings)} semantic, {len(ed.value_embeddings)} value vectors")
print(f"\n{len(embedded)} dictionaries embedded.")

INFO: No device provided, using mps

INFO: Loading SentenceTransformer model from sentence-transformers/all-mpnet-base-v2.

/tmp/ddh_release_venv/lib/python3.12/site-packages/ddharmon/embedding/provider.py:77: FutureWarning: The `get_sentence_embedding_dimension` method has been renamed to `get_embedding_dimension`.
  self._dimension: int = self._model.get_sentence_embedding_dimension()  # type: ignore[assignment]
INFO: semantic embeddings: 0 cached, 22265 new

Embedding model loaded: all-mpnet-base-v2 (768d)

Batches:   0%|          | 0/348 [00:00<?, ?it/s]

Batches:   0%|          | 1/348 [00:04<26:12,  4.53s/it]

Batches:   1%|          | 2/348 [00:06<15:53,  2.76s/it]

WORK = Path("harmonization_artifacts")
WORK.mkdir(exist_ok=True)


def classify_via_batch(records):
    """Classify A/R/N prompts via the Anthropic Batch API. Returns {id: response}."""
    if not records:
        return {}
    prompts_path = WORK / "prompts_harmonize_arn.jsonl"
    responses_path = WORK / "responses_harmonize_arn.jsonl"
    write_prompts_jsonl(records, prompts_path)
    submit_and_wait(prompts_path, responses_path)  # needs ANTHROPIC_API_KEY
    responses = {}
    with open(responses_path) as f:
        for line in f:
            rec = json.loads(line)
            responses[rec["id"]] = rec["response"]
    return responses


result = harmonize_dictionaries(embedded, classify=classify_via_batch, min_cluster_size=15)
print(f"{len(result.verdicts)} sub-cluster verdicts ({len(result.prompt_records)} via LLM)")

INFO: topic_model_dictionaries: 29173 fields from 3 dicts

INFO: topic_model_dictionaries complete: 29173 fields, 516 topics, 7249 outliers in 549.49s

INFO: prepare_from_clusters: 516 clusters -> 10 LLM prompts, 1634 deterministic verdicts

INFO: Submitting 10 requests to Batch API (models=['claude-sonnet-4-6'])

INFO: Batch submitted: msgbatch_015SiAgS36VmYs3h57Rj3GqH (manifest: harmonization_artifacts/prompts_harmonize_arn.jsonl.batch_manifest.json)

Submitted batch msgbatch_015SiAgS36VmYs3h57Rj3GqH; polling every 60s until it ends...

  in_progress: succeeded=0 processing=10 errored=0

  in_progress: succeeded=0 processing=10 errored=0

  in_progress: succeeded=0 processing=10 errored=0

INFO: Batch msgbatch_015SiAgS36VmYs3h57Rj3GqH: status=ended, succeeded=10, processing=0, errored=0, expired=0, canceled=0

counts = write_buckets(result, WORK / "buckets")
n_eitl = export_eitl_queue(result, WORK / "eitl_queue.tsv")
print("Buckets:", counts)
print(f"EITL queue: {n_eitl} sub-clusters -> {WORK / 'eitl_queue.tsv'}")

Buckets: {'cde_only': 1018, 'noise': 255, 'single_cohort': 355, 'novel': 11, 'adopt': 1, 'unaligned': 3, 'refine': 1}
EITL queue: 371 sub-clusters -> harmonization_artifacts/eitl_queue.tsv

import pandas as pd

df = pd.DataFrame([
    {
        "sub_cluster": v.sub_cluster_id,
        "label": v.label,
        "verdict": v.verdict or "(skipped)",
        "mode": v.mode,
        "parent_cde_id": v.parent_cde_id,
        "anchor": v.anchor_designation,
        "confidence": v.confidence,
        "n_fields": v.n_fields,
        "cohorts": ";".join(v.cohorts),
        "decided_by": v.decided_by,
    }
    for v in result.verdicts
])
print(df["verdict"].value_counts())
df.sort_values(["verdict", "confidence"], ascending=[True, False]).head(30)

verdict
(skipped)    1628
novel          11
unaligned       3
adopt           1
refine          1
Name: count, dtype: int64

	sub_cluster	label	verdict	mode	parent_cde_id	anchor	confidence	n_fields	cohorts	decided_by
0	0:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
1	0:1		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
2	0:-1		(skipped)	noise	NaN	NaN	NaN	0		deterministic
3	0:3		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
4	0:2		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
5	0:4		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
6	1:-1		(skipped)	noise	NaN	NaN	NaN	0		deterministic
7	1:1		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
8	1:2		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
9	1:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
10	2:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
11	2:1		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
12	2:2		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
13	2:-1		(skipped)	noise	NaN	NaN	NaN	0		deterministic
14	3:3		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
15	3:1		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
16	3:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
17	3:4		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
18	3:-1		(skipped)	noise	NaN	NaN	NaN	0		deterministic
19	3:2		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
20	4:-1		(skipped)	noise	NaN	NaN	NaN	0		deterministic
21	4:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
22	4:3		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
23	4:5		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
24	4:1		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
25	4:4		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
26	4:7		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
27	4:6		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic
28	4:2	Last / Months / Com	(skipped)	single_cohort	NaN	NaN	NaN	12	CLSA	deterministic
29	5:0		(skipped)	cde_only	NaN	NaN	NaN	0		deterministic

ddharmon v1 — Sub-cluster-anchored CDE Harmonization¶

1. Load cohorts + CDE catalog¶

2. Embed — dual vectors (semantic + value)¶

3–6. Run the pipeline (cluster → sub-cluster → anchor → classify)¶

7. Outputs → buckets + EITL review queue¶

Inspect verdicts¶