This work began as an exploration of Graph RAG for medical literature. It quickly became clear that the needs of researchers and clinicians marked this as a "high stakes" area of reasoning and retrieval, and that this would be an important consideration in every decision in the design of a suitable system. Specifically three principles emerged as significant requirements.
- The knowledge graph should use types and type-checking tools as a guard against meaningless contents. These aid in the ingestion of source documents, offering guidance in parsing ambiguous input.
- The medical field has accumulated a number of authoritative ontologies for diseases, drugs, genes, biological organisms, and other things discussed in the literature. Using ontology references as canonical IDs in the graph enables the system to make comparisons from one paper to another, and to merge graphs from different sources, with some confidence that apples will be compared to apples.
- The provenance of claims, findings, and assertions should be meticulously tracked thru the graph in order that the validity of chains if reasoning can be reviewed. In a medical setting, nobody wants to find themselves in doubt as to whether or not a conclusion is the result of LLM hallucination. All conclusions must be traceable back to the source documents used to arrive at those conclusions, including any steps of deduction
While my primary early interest was (and remains) in medical literature, I thought it a good idea to exercise the system in a very different domain just to kick the tires. I was particularly curious about machine reasoning, and had been for a long time, and decided that a useful exercise would be to use a knowledge graph to "solve" a Sherlock Holmes mystery, to have it arrive at a solution computationally, given the contents of the story right up to (but not including) Holmes' announcement of the solution.
The Holmes graph has some very different parameters. Provenance doesn't really matter. For canonical IDs we don't find the kinds of ontologies that biologists and medical researchers have curated over decades, and for my own work I went with the fan-curated Baker Street Wiki. The typed graph was useful in guiding the process of parsing text into meaningful graph contents.
One surprising result of this work is that I learned things about Sherlock Holmes's reasoning process that I didn't know before. I was suprised to learn that he was comfortable working with varying levels of uncertainty. Holmes famously said, "I never guess. It is a shocking habit -- destructive to the logical faculty" and this led me to imagine that he would insist on dealing only with certainties. In the story I chose, "A Scandal in Bohemia", his solution is the likeliest of a short list of possibilities.
https://github.com/graphwright/ner-20260608
A remarkable thing about Pydantic models is that you can easily get them to print themselves in a form that can be imported back into a Python interpreter directly.
Some of the most fundamental pieces of the graph look like this.
Here's one approach
class TruthStatus(str, Enum):
ASSERTED_TRUE = "asserted_true"
ASSERTED_FALSE = "asserted_false"
HYPOTHETICAL = "hypothetical"
DISPUTED = "disputed"
RETRACTED = "retracted"
class Trait:
"""Marker base for all semantic traits."""
class Transitive(Trait): ...
class Symmetric(Trait): ...
class Functional(Trait): ...
class InverseFunctional(Trait): ...
class EntityInstance(BaseModel):
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
id: str
def __repr__(self) -> str:
return f"{type(self).__name__}({self.id!r})"
def __str__(self) -> str:
return self.id
def statement_id(subject_id: str, predicate_name: str, object_id: str) -> str:
"""Content-addressed id for a predicate instance.
The predicate name and participant ids are embedded for traceability and
idempotent construction — not for type dispatch. This is distinct from the
discouraged synthetic-entity pattern (sib:event:kings_visit): here the
human-readable components serve a functional purpose (deduplication, readable
debugging) and nothing in the system parses the string to recover a type.
"""
return f"stmt:{subject_id}:{predicate_name}:{object_id}"
class BaseStatement(EntityInstance):
truth_status: TruthStatus = TruthStatus.HYPOTHETICAL
def __str__(self) -> str:
subj = getattr(self, 'subject', None)
obj = getattr(self, 'object_', None)
cls = type(self).__name__
if subj is not None and obj is not None:
return f"{cls}({subj} → {obj})"
return clsHere's another
# truth_status vocabulary from formal-defns.md
TruthStatus = Literal[
"asserted_true",
"asserted_false",
"hypothetical",
"disputed",
"retracted",
]
# How a claim was derived. A closed vocabulary (like TruthStatus) so values do not
# drift into "inferred" / "inference" / "derived" variants.
ExtractionMethod = Literal[
"manual",
"inferred",
"quotation",
"model_extraction",
]
class Provenance(BaseModel):
"""One provenance record describing how a statement was produced."""
model_config = ConfigDict(frozen=True)
source: str
extraction_method: ExtractionMethod
class Instance(BaseModel):
"""A member of V. Common root of entity instances and statements."""
model_config = ConfigDict(frozen=True)
id: str
class EntityInstance(Instance):
"""A member of V whose type is an entity type. Entity types are
subclasses of this."""
SubjectT = TypeVar("SubjectT", bound=EntityInstance)
ObjectT = TypeVar("ObjectT", bound=Instance)
class BaseStatement(Instance, Generic[SubjectT, ObjectT]):
"""A predicate instance: a member of V and of the derived edge set E
(E subset of V, since BaseStatement inherits Instance).
Domain and range are sets of types, dom(p) and ran(p). Each is expressed as
the single type parameter SubjectT / ObjectT, bound to a Union when the set
has more than one member -- e.g. BaseStatement[Person | Organization, Vehicle]
means dom = {Person, Organization}, ran = {Vehicle}. A singleton set is just
one type. There is no custom validation; mypy enforces membership statically
and Pydantic at construction time.
ObjectT is bound to Instance (not EntityInstance) so a statement's object may
itself be a statement (higher-order predication). For a range of "any
statement", use `AnyStatement` (below) rather than a bare or Any-parametrized
BaseStatement, so the object's concrete predicate type is preserved.
`provenance` is stored as `tuple[Provenance, ...] | None`. For convenience,
callers may pass a single `Provenance` or a `list[Provenance]`; the validator
normalizes either form to a tuple before model construction.
"""
subject: SubjectT
object_: ObjectT
truth_status: TruthStatus = "hypothetical"
provenance: tuple[Provenance, ...] | None = None
@model_validator(mode="before")
@classmethod
def _ensure_provenance_tuple(cls, data: Dict[str, Any]) -> Any:
"""Normalize single or list provenance inputs to a tuple."""
# if not isinstance(data, dict):
# return data
provenance = data.get("provenance")
if provenance is None or isinstance(provenance, tuple):
return data
normalized = dict(data)
if isinstance(provenance, list):
normalized["provenance"] = tuple(cast(list[Provenance], provenance))
else:
normalized["provenance"] = (provenance,)
return normalized
@model_validator(mode="after")
def _reject_empty_provenance(self) -> Self:
if self.provenance == ():
raise ValueError(
"provenance must contain at least one record when present; "
"use None for ungrounded statements"
)
return self
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
# Subscripting BaseStatement[...] makes Pydantic create generic submodels
# while this module is still loading (e.g. the PartnerT bound below), which
# fires this hook before the validator is defined. Skip until it exists;
# real domain subclasses are always created after base.py finishes loading.
validate = globals().get("_validate_inverse_declaration")
if validate is not None:
validate(cls)
# An object range meaning "any statement" (higher-order predication). Use this
# rather than a bare or Any-parametrized BaseStatement: InstanceOf validates by
# isinstance and keeps the value as-is, so the object's concrete predicate type
# (tau) is preserved. A parametrized `BaseStatement[Any, Any]` range would
# instead rebuild the object as the base class, discarding its type. Trade-off:
# an object supplied as a raw dict is rejected -- reconstruct statements via the
# loader and pass instances, which is what the serialization design does anyway.
AnyStatement = InstanceOf[BaseStatement[Any, Any]]
# Shorthand for "a statement of any subject/object type", for annotations that
# would otherwise repeat BaseStatement[Any, Any] (and its type: ignore[type-arg]).
AnyStmt: TypeAlias = BaseStatement[Any, Any]
# ---------------------------------------------------------------------------
# Traits: declarative semantic properties of a predicate *type* (R1).
#
# A trait belongs to the predicate type, never to an individual statement. It
# is realized as a mixin class inherited alongside a BaseStatement subclass:
#
# class Knows(BaseStatement[Person, Person], Symmetric): ...
#
# The unparameterized traits are plain markers, introspectable with
# issubclass(). Inverse is generic, parameterized by the partner predicate
# type and resolved with get_inverse(). Rule (the Datalog escape hatch) has no
# clean type-level expression, so it is declared in prose on the predicate
# class and realized as a callable the inference engine invokes -- not here.
# ---------------------------------------------------------------------------
class Symmetric:
"""p(x, y) implies p(y, x)."""
class Transitive:
"""p(x, y) and p(y, z) imply p(x, z)."""
class Functional:
"""Each subject has at most one object under p."""
class InverseFunctional:
"""Each object has at most one subject under p."""
PartnerT = TypeVar("PartnerT", bound=BaseStatement[Any, Any])
class Inverse(Generic[PartnerT]):
"""p(x, y) implies p'(y, x), where the partner predicate p' is supplied as
the type argument -- e.g. class ChildOf(BaseStatement[...], Inverse[ParentOf]).
Declared one-way: `get_inverse(ChildOf)` is `ParentOf`, but
`get_inverse(ParentOf)` is `None` unless `ParentOf` also declares the inverse.
The declaring class's domain/range must be the swap of the partner's; a
mismatch is a TypeError at import (see `_validate_inverse_declaration`)."""
def get_inverse(
stmt_type: type[BaseStatement[Any, Any]],
) -> type[BaseStatement[Any, Any]] | None:
"""Return the partner predicate type if `stmt_type` declares Inverse[...],
else None.
Resolves the type argument from the declared bases. A partner given as a
forward reference -- `Inverse["ParentOf"]`, needed when the partner is
defined later or the two predicates are mutual inverses -- is resolved
against the declaring module's globals. This is runtime introspection of the
schema; it does not touch or revalidate instances.
"""
for base in getattr(stmt_type, "__orig_bases__", ()):
if get_origin(base) is Inverse:
(partner,) = get_args(base)
if isinstance(partner, ForwardRef):
partner = partner.__forward_arg__
if isinstance(partner, str):
module = sys.modules.get(stmt_type.__module__)
resolved = getattr(module, partner, None)
if resolved is None:
raise NameError(
f"Inverse partner {partner!r} of {stmt_type.__name__} is "
f"not resolvable in module {stmt_type.__module__!r}"
)
partner = resolved
return cast(type[BaseStatement[Any, Any]], partner)
return None
def _tn(t: Any) -> str:
"""A display name for a type or type expression (handles unions with no __name__)."""
return getattr(t, "__name__", repr(t))
---
class SherlockEntity(EntityInstance):
"""Base entity with optional source metadata from the JSONL catalog."""
canonical: str
aliases: tuple[str, ...] = ()
wiki_url: str | None = None
raw_type: str | None = None
class Person(SherlockEntity):
"""A person in the story world."""
class Organization(SherlockEntity):
"""An organization in the story world."""
class Location(SherlockEntity):
"""A place/location in the story world."""
class Object(SherlockEntity):
"""A tangible or conceptual object in the story world."""
class Event(SherlockEntity):
"""An event node imported from event extraction."""
class Moment(SherlockEntity):
"""A time/moment node imported from timeline extraction."""
class OtherEntity(SherlockEntity):
"""Fallback entity type when the source type is unknown."""
SubjectT = TypeVar("SubjectT", bound=SherlockEntity)
ObjectT = TypeVar("ObjectT", bound=SherlockEntity)
class StoryStatement(BaseStatement[SubjectT, ObjectT], Generic[SubjectT, ObjectT]):
"""Statement enriched with story-extraction metadata from triplet rows."""
story_id: str
paragraph_index: int | None = None
sentence_ids: tuple[int, ...] = ()
asserting_narrator_id: str | None = None
extraction_confidence: float | None = None
narrator_confidence: float | None = None
raw_extraction_method: str | None = None
class Involves(StoryStatement[Event, Person]):
"""An event involves a person."""
class OccurredAt(StoryStatement[Event, Moment]):
"""An event occurred at a specific moment."""
class Possesses(StoryStatement[Person, Object]):
"""A person possesses an object."""
class AssociatedWith(StoryStatement[Person, Location]):
"""A person is associated with a location."""
class Knows(StoryStatement[Person, Person], Symmetric):
"""A social knowledge relation between two people."""
class LocatedIn(StoryStatement[Location, Location], Transitive):
"""A transitive containment/location relation."""
class HappenedIn(StoryStatement[Event, Location]):
"""An event took place in a location.
This captures event->place structure that is often only implicit in event
identifiers/descriptions from extraction output.
"""
class PhysicallyIn(BaseStatement[Object, Location]):
"""An object is physically located in a place.
This is a *derivable* predicate, not an extracted one: it is produced by
inference (e.g. the mystery Horn clause), never imported from a triplet row.
It therefore subclasses ``BaseStatement`` directly rather than
``StoryStatement`` -- an inferred fact has no story-extraction metadata
(no ``story_id``, paragraph index, or extraction confidence), and requiring
those fields would make it impossible for the datalog engine to construct a
derived head. Keeping extracted and inferred predicates on separate branches
of the hierarchy is the honest ontology: provenance-of-extraction belongs
only to things that were extracted.
"""
---
entity_arnsworth_castle_business = OtherEntity(
id="entity:arnsworth_castle_business",
canonical="Arnsworth Castle business",
aliases=("Arnsworth Castle business",),
wiki_url=None,
raw_type="other",
)
entity_atkinson_brothers = OtherEntity(
id="entity:atkinson_brothers",
canonical="Atkinson brothers",
aliases=("Atkinson brothers",),
wiki_url=None,
raw_type="other",
)
entity_darlington_substitution_scandal = OtherEntity(
id="entity:darlington_substitution_scandal",
canonical="Darlington Substitution Scandal",
aliases=("Darlington Substitution Scandal",),
wiki_url=None,
raw_type="other",
)
entity_trepoff_murder = OtherEntity(
id="entity:trepoff_murder",
canonical="Trepoff murder",
aliases=("Trepoff murder",),
wiki_url=None,
raw_type="other",
)
obj_5_15_train = Object(
id="obj:5_15_train",
canonical="5:15 train",
aliases=("5:15 train",),
wiki_url=None,
raw_type="object",
)
obj_a_study_in_scarlet = Object(
id="obj:a_study_in_scarlet",
canonical="A Study in Scarlet",
aliases=("A Study in Scarlet",),
wiki_url=None,
raw_type="object",
)
obj_armchair = Object(
id="obj:armchair",
canonical="armchair",
aliases=("armchair", "the chair"),
wiki_url=None,
raw_type="object",
)
obj_bell_pull = Object(
id="obj:bell-pull",
canonical="bell-pull",
aliases=("bell-pull", "the right bell-pull"),
wiki_url=None,
raw_type="object",
)
obj_black_vizard_mask = Object(
id="obj:black_vizard_mask",
canonical="black vizard mask",
aliases=("black vizard mask", "mask"),
wiki_url=None,
raw_type="object",
)
obj_blinds = Object(
id="obj:blinds",
canonical="blinds",
aliases=("blinds",),
wiki_url=None,
raw_type="object",
)
obj_bohemian_paper = Object(
id="obj:bohemian_paper",
canonical="Bohemian paper",
aliases=("Bohemian paper", "private note-paper", "the note paper", "the notepaper"),
wiki_url=None,
raw_type="object",
)
obj_brooch_with_flaming_beryl = Object(
id="obj:brooch_with_flaming_beryl",
canonical="brooch with flaming beryl",
aliases=("brooch with flaming beryl",),
wiki_url=None,
raw_type="object",
)
obj_cab = Object(
id="obj:cab",
canonical="cab",
aliases=("cab", "the hansom cab"),
wiki_url=None,
raw_type="object",
)
obj_cap = Object(
id="obj:cap", canonical="cap", aliases=("cap",), wiki_url=None, raw_type="object"
)
obj_chamois_leather_bag = Object(
id="obj:chamois_leather_bag",
canonical="chamois leather bag",
aliases=("chamois leather bag",),
wiki_url=None,
raw_type="object",
)
obj_chubb_lock = Object(
id="obj:chubb_lock",
canonical="Chubb lock",
aliases=("Chubb lock",),
wiki_url=None,
raw_type="object",
)
obj_cigar_case = Object(
id="obj:cigar_case",
canonical="cigar case",
aliases=("cigar case",),
wiki_url=None,
raw_type="object",
)
obj_cold_beef_and_beer = Object(
id="obj:cold_beef_and_beer",
canonical="cold beef and beer",
aliases=("cold beef and beer",),
wiki_url=None,
raw_type="object",
)
obj_continental_gazetteer = Object(
id="obj:continental_gazetteer",
canonical="Continental Gazetteer",
aliases=("Continental Gazetteer",),
wiki_url=None,
raw_type="object",
)
...
person_nurse_girl = Person(
id="person:nurse-girl",
canonical="nurse-girl",
aliases=("nurse-girl",),
wiki_url=None,
raw_type="person",
)
person_staff_commander = Person(
id="person:staff-commander",
canonical="staff-commander",
aliases=("staff-commander",),
wiki_url=None,
raw_type="person",
)
person_woman_bystander = Person(
id="person:woman_bystander",
canonical="woman bystander",
aliases=("woman bystander",),
wiki_url=None,
raw_type="person",
)
place_baker_street = Location(
id="place:baker_street",
canonical="Baker Street",
aliases=(
"Baker Street",
"Holmes's chambers",
"Holmes's rooms",
"the chamber (Baker Street rooms)",
"steps from hall to Holmes's room",
"the seventeen steps",
),
wiki_url=None,
raw_type="place",
)
place_carlsbad = Location(
id="place:carlsbad",
canonical="Carlsbad",
aliases=("Carlsbad",),
wiki_url=None,
raw_type="place",
)