import hpotk
import pandas as pd
import tqdm
import re
import typing
import json
import numpy as np
from fastembed import TextEmbedding
from sentence_transformers import SentenceTransformer
store = hpotk.configure_ontology_store()
hpo = store.load_hpo()HPO Vector Database
Adapted from the code for HPO-RAG, 2025, Genome Med
To run the notebook, install the following packages
pip install hpo-toolkit
pip install fastembed
pip install sentence_transformers1. Download the latest HPO (hp.json)
2. Collect all HPO terms
We use the HPO toolkit, a Python package for working with HPO. The following code includes several coversions between TermId objects and the corresponding strings.
ROOT_ID = hpotk.TermId.from_curie('HP:0000001') # “All”
PHENO_ID = hpotk.TermId.from_curie('HP:0000118') # “Phenotypic abnormality”
def _to_curie(term_id) -> str:
"""Normalize an HPO id-like object/string to the canonical 'HP:NNNNNNN' form."""
s = str(term_id)
if "_" in s and ":" not in s:
s = s.replace("_", ":", 1)
return s
ROOT_STR = _to_curie(ROOT_ID)
PHENO_STR = _to_curie(PHENO_ID)
hpo_terms = dict()
root_term = hpo.get_term(ROOT_ID)
hpo_terms[_to_curie(ROOT_ID)] = root_term
pheno_root_term = hpo.get_term(PHENO_ID)
hpo_terms[_to_curie(PHENO_ID)] = pheno_root_term
for term_id in hpo.graph.get_descendants(ROOT_ID):
term = hpo.get_term(term_id)
if not term:
print(f"Error-could not retrieve term for {term_id}")
continue
hpo_terms[_to_curie(term_id)] = term
print(f"Got a total of {len(hpo_terms)} HPO terms.")Got a total of 19894 HPO terms.Create a map that has the parents of each term
parent_map = {
_to_curie(term_id): [_to_curie(p) for p in hpo.graph.get_parents(term_id)]
for term_id in hpo.graph.get_descendants(ROOT_ID)
}
label_map = {_to_curie(term_id): term.name for term_id, term in hpo_terms.items()}Multi-Path Lineage Helper
Create a map of paths from the root of the HPO to each term
_lineage_memo = {}
def _build_lineage_paths(hp_id:hpotk.TermId, parent_map, seen=None):
"""
Return all paths from ROOT_ID → ... → hp_id.
Each path is a list of HP_ID strings.
Avoids cycles by tracking `seen`.
"""
if seen is None:
seen = set()
if hp_id in seen: # cycle guard
return []
seen.add(hp_id)
if hp_id in _lineage_memo:
return _lineage_memo[hp_id]
# base case: we hit the root term
if hp_id == ROOT_STR:
paths = [[ROOT_STR]]
else:
parents = parent_map.get(hp_id, [])
if not parents:
# orphan: just attach root + self
paths = [[_to_curie(ROOT_STR), hp_id]]
else:
paths = []
for p in parents:
for ppath in _build_lineage_paths(p, parent_map, seen):
paths.append(ppath + [hp_id])
_lineage_memo[hp_id] = paths
return pathsbuild_hpo_dataframe
This function arranges the data retrieved for each HPO term as a pandas dataframe.
CLEAN_ABNORMALITY = re.compile(r'(?i)^Abnormality of(?: the)?\s*')
def build_hpo_dataframe(hpo_terms:typing.Dict[str,hpotk.Term]) -> pd.DataFrame:
records = []
terms = list(hpo_terms.values())
for term in tqdm.tqdm(terms, desc="Building HPO DataFrame", unit="term"):
hp_id = _to_curie(term.identifier)
label = term.name
definition = term.definition or ""
synonyms = [syn.name for syn in (term.synonyms or [])]
#ALT IDs : omitting for this demo!
# XREFS: omitting for this demo!
# ── LINEAGE ──
paths = _build_lineage_paths(hp_id, parent_map) or [[_to_curie(ROOT_ID), hp_id]]
PHENO_ID_str = _to_curie(PHENO_ID)
for path_ids in paths:
lineage_str = " -> ".join(f"{label_map[_to_curie(i)]} ({_to_curie(i)})" for i in path_ids)
# ── ORGAN SYSTEM ──
if PHENO_ID_str in path_ids:
idx = path_ids.index(PHENO_ID_str)
organ = label_map.get(path_ids[idx+1], "Other") if idx+1 < len(path_ids) else "Other"
else:
organ = "Other"
organ_system = CLEAN_ABNORMALITY.sub("", organ).title()
# ── EMIT ROWS ──
for phrase in [label] + synonyms:
if not phrase:
continue
# **Only title-case** the phrase; do NOT strip anything from it
clean_phrase = phrase.title()
records.append({
"hp_id": _to_curie(hp_id),
"phrase": clean_phrase,
"organ_system": organ_system,
"lineage": lineage_str,
"definition": definition,
})
return pd.DataFrame(
records,
columns=[
"hp_id", "phrase", "organ_system", "lineage","definition"
]
)hpo_df = build_hpo_dataframe(hpo_terms=hpo_terms)
hpo_df.head(10)Building HPO DataFrame: 100%|██████████| 19894/19894 [00:00<00:00, 70143.65term/s]| hp_id | phrase | organ_system | lineage | definition | |
|---|---|---|---|---|---|
| 0 | HP:0000001 | All | Other | All (HP:0000001) | |
| 1 | HP:0000118 | Phenotypic Abnormality | Other | All (HP:0000001) -> Phenotypic abnormality (HP... | Definition(definition="A phenotypic abnormalit... |
| 2 | HP:0000118 | Organ Abnormality | Other | All (HP:0000001) -> Phenotypic abnormality (HP... | Definition(definition="A phenotypic abnormalit... |
| 3 | HP:0040279 | Frequency | Other | All (HP:0000001) -> Frequency (HP:0040279) | Definition(definition="Class to represent freq... |
| 4 | HP:0040285 | Excluded | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 0% of the ca... |
| 5 | HP:0040285 | Excluded (0%) | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 0% of the ca... |
| 6 | HP:0040284 | Very Rare | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 1% to 4% of ... |
| 7 | HP:0040284 | Very Rare (≪4-1%) | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 1% to 4% of ... |
| 8 | HP:0040284 | Very Rare (<4-1%) | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 1% to 4% of ... |
| 9 | HP:0040283 | Occasional | Other | All (HP:0000001) -> Frequency (HP:0040279) -> ... | Definition(definition="Present in 5% to 29% of... |
Embedding model
PAT = re.compile(r'\s*\([^)]*\)\s*')
def clean_text(txt: str) -> str:
txt = PAT.sub(' ', txt)
txt = re.sub(r'\s+', ' ', txt).strip().lower()
txt = re.sub(r'[^\w\s]+$', '', txt)
return txt
def get_embedding_model(
use_sbert: bool = True,
sbert_model: str = 'pritamdeka/SapBERT-mnli-snli-scinli-scitail-mednli-stsb',
bge_model: str = 'BAAI/bge-small-en-v1.5',
):
if use_sbert:
print(f"Loading SBERT model: {sbert_model}")
return SentenceTransformer(sbert_model)
print(f"Loading BGE model: {bge_model}")
return TextEmbedding(model_name=bge_model)/Users/peterrobinson/GIT/bio-ontollm-hm/.venv/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
from .autonotebook import tqdm as notebook_tqdm5. Vectorize & Save
def vectorize_dataframe(
df: pd.DataFrame,
meta_out: str,
vec_out: str,
use_sbert: bool = True
):
model = get_embedding_model(use_sbert)
constants: dict[str, dict] = {}
entries: list[dict] = []
embs: list[np.ndarray] = []
NEG_PATTERN = re.compile(
r'\b(?:decreas(?:e|ed|ing)?|loss(?:es)?|hypo[-]?\w+)\b',
re.IGNORECASE
)
POS_PATTERN = re.compile(
r'\b(?:increas(?:e|ed|ing)?|gain(?:s|ed)?|hyper[-]?\w+)\b',
re.IGNORECASE
)
for _, row in tqdm.tqdm(df.iterrows(), total=len(df), desc="Embedding rows", unit="row"):
hp_id = str(row.hp_id) # convert from HPOTK TermId object
info = clean_text(row.phrase)
direction = 0
if NEG_PATTERN.search(info):
direction = -1
elif POS_PATTERN.search(info):
direction = 1
# ——— Embedding call unchanged ———
if use_sbert:
vec = model.encode(info, convert_to_numpy=True)
else:
vec = np.asarray(list(model.embed([info]))[0], dtype=np.float32)
entries.append({
'hp_id': hp_id,
'info': info,
'direction':direction
})
if hp_id not in constants:
constants[hp_id] = {
'organ_system': row.organ_system,
'lineage': row.lineage,
'definition': getattr(row.definition, 'definition', "") or "",
}
embs.append(vec.astype(np.float16))
# ——— Save embeddings ———
emb_matrix = np.vstack(embs)
combined = {
'constants': constants,
'entries': entries
}
with open(meta_out, 'w') as f:
json.dump(combined, f, separators=(',', ':'))
np.savez_compressed(vec_out, emb=emb_matrix)
print(f"Saved {len(entries)} embeddings → {meta_out}, {vec_out}")hpo_df.to_csv("hpo_terms_full.csv", index=False)
print(f"Built DataFrame with {len(hpo_df)} rows.")
vectorize_dataframe(
hpo_df,
meta_out='hpo_meta.json',
vec_out='hpo_embedded.npz',
use_sbert=True
)Warning: You are sending unauthenticated requests to the HF Hub. Please set a HF_TOKEN to enable higher rate limits and faster downloads.Built DataFrame with 143238 rows.
Loading SBERT model: pritamdeka/SapBERT-mnli-snli-scinli-scitail-mednli-stsbLoading weights: 100%|██████████| 199/199 [00:00<00:00, 68426.50it/s]
Embedding rows: 100%|██████████| 143238/143238 [15:30<00:00, 153.89row/s]Saved 143238 embeddings → hpo_meta.json, hpo_embedded.npzwith open('hpo_meta.json') as f:
meta = json.load(f)
print(f"Total entries: {len(meta['entries'])}")
print(f"Total unique HP terms (constants): {len(meta['constants'])}")
for entry in meta['entries'][1000:1005]:
hp_id = entry['hp_id']
const = meta['constants'][hp_id]
print(f"HP ID: {hp_id}")
print(f"Info: {entry['info']}")
print(f"Direction: {entry['direction']}")
print(f"Organ system:{const['organ_system']}")
print(f"Lineage: {const['lineage']}")
print(f"Definition: {const['definition']}")
print("-" * 60)Total entries: 143238
Total unique HP terms (constants): 19894
HP ID: HP:0025212
Info: triggered by fasting
Direction: 0
Organ system:Other
Lineage: All (HP:0000001) -> Clinical modifier (HP:0012823) -> Triggered by (HP:0025204) -> Triggered by fasting (HP:0025212)
Definition: Applies to a sign or symptom that is provoked or brought about by abstaining from eating food (fasting).
------------------------------------------------------------
HP ID: HP:0025207
Info: triggered by dehydration
Direction: 0
Organ system:Other
Lineage: All (HP:0000001) -> Clinical modifier (HP:0012823) -> Triggered by (HP:0025204) -> Triggered by dehydration (HP:0025207)
Definition: Applies to a sign or symptom that is provoked or brought about by being dehydrated, i.e., by a deficit in total body water.
------------------------------------------------------------
HP ID: HP:0025207
Info: dehydration triggered symptoms
Direction: 0
Organ system:Other
Lineage: All (HP:0000001) -> Clinical modifier (HP:0012823) -> Triggered by (HP:0025204) -> Triggered by dehydration (HP:0025207)
Definition: Applies to a sign or symptom that is provoked or brought about by being dehydrated, i.e., by a deficit in total body water.
------------------------------------------------------------
HP ID: HP:0025207
Info: triggered by dehydration
Direction: 0
Organ system:Other
Lineage: All (HP:0000001) -> Clinical modifier (HP:0012823) -> Triggered by (HP:0025204) -> Triggered by dehydration (HP:0025207)
Definition: Applies to a sign or symptom that is provoked or brought about by being dehydrated, i.e., by a deficit in total body water.
------------------------------------------------------------
HP ID: HP:0025206
Info: triggered by cold
Direction: 0
Organ system:Other
Lineage: All (HP:0000001) -> Clinical modifier (HP:0012823) -> Triggered by (HP:0025204) -> Triggered by cold (HP:0025206)
Definition: Applies to a sign or symptom that is provoked or brought about by exposure to cold surroundings.
------------------------------------------------------------print(f"parent_map size: {len(parent_map)}")
print(f"parent_map sample keys: {list(parent_map.keys())[:5]}")
print(f"parent_map sample value: {list(parent_map.items())[:1]}")
test_id = 'HP:0045027'
print(f"test_id: {test_id!r}, type: {type(test_id)}")
print(f"test_id in parent_map: {test_id in parent_map}")
print(_build_lineage_paths(test_id, parent_map))parent_map size: 19893
parent_map sample keys: ['HP:0040279', 'HP:0040285', 'HP:0040284', 'HP:0040283', 'HP:0040282']
parent_map sample value: [('HP:0040279', ['HP:0000001'])]
test_id: 'HP:0045027', type: <class 'str'>
test_id in parent_map: True
[['HP:0000001', 'HP:0000118', 'HP:0045027']]