semantica/cookbook/advanced/14_Datalog_Style_Reasoning.ipynb

Datalog-Style Reasoning

End-to-end guide to Semantica's DatalogReasoner — a native bottom-up semi-naive fixpoint engine — wired together with GraphBuilder, ContextGraph, GraphAnalyzer, ExplanationGenerator, and the supporting data-classes (DatalogFact, DatalogRule, InferenceResult, Rule).

What you will build

Part	Topic	Key classes
1	Core API & EDB/IDB concepts	DatalogReasoner, DatalogFact, DatalogRule
2	KG → Datalog pipeline	GraphBuilder, GraphAnalyzer, DatalogReasoner
3	ContextGraph integration	ContextGraph, DatalogReasoner.load_from_graph()
4	RBAC access-control policy	GraphBuilder, DatalogReasoner, ExplanationGenerator
5	Org hierarchy	ContextGraph, DatalogReasoner, InferenceResult
6	Engine introspection	DatalogFact, DatalogRule internal state

Related notebooks

• 08_Reasoning_and_Inference.ipynb — high-level Reasoner with IF/THEN syntax
• 10_Temporal_Knowledge_Graphs.ipynb — temporal reasoning

Documentation: Reasoning API | KG API | Context API

In [ ]:

!pip install -qU semantica

In [ ]:

# ── Reasoning ──────────────────────────────────────────────────────────────
from semantica.reasoning import (
    DatalogReasoner,   # native Datalog fixpoint engine
    DatalogFact,       # frozen dataclass: predicate + args tuple
    DatalogRule,       # dataclass: head + body (list[BodyAtom])
    ExplanationGenerator,  # generates NL justifications
    InferenceResult,   # result dataclass consumed by ExplanationGenerator
    Rule,              # rule dataclass used by ExplanationGenerator
    RuleType,          # enum: IMPLICATION | EQUIVALENCE | CONSTRAINT | TRANSFORMATION
)

# ── Knowledge Graph ────────────────────────────────────────────────────────
from semantica.kg import (
    GraphBuilder,    # constructs KG dicts from entity+relationship sources
    GraphAnalyzer,   # centrality, communities, connectivity, metrics
)

# ── Context ────────────────────────────────────────────────────────────────
from semantica.context import ContextGraph  # in-memory graph: add_node/add_edge/find_*

print("All Semantica classes imported successfully.")

---

Part 1 — Core API: EDB Facts, IDB Rules, Fixpoint

Datalog in 30 seconds

Term	Meaning	Example
EDB (Extensional DB)	Ground facts you assert	parent(tom, bob)
IDB (Intensional DB)	Facts derived by rules	ancestor(tom, ann)
Rule (Horn clause)	If body → derive head	ancestor(X,Y) :- parent(X,Y).
Variable	Uppercase, unified during eval	X, Y, Role
Constant	Lowercase, matches literally	tom, admin
Fixpoint	Iterate until no new facts appear	DatalogReasoner.derive_all()

The canonical example — transitive ancestry

In [ ]:

# ── Step 1: create engine ──────────────────────────────────────────────────
dr = DatalogReasoner()

# ── Step 2: load EDB (ground facts) ───────────────────────────────────────
# Syntax: predicate(constant1, constant2)  — constants must be lowercase
edb_facts = [
    "parent(tom, bob)",
    "parent(bob, ann)",
    "parent(ann, pat)",
]
for f in edb_facts:
    dr.add_fact(f)

print(f"EDB loaded: {len(dr._all_facts)} ground facts")

In [ ]:

# ── Step 3: add IDB rules (Horn clauses) ──────────────────────────────────
# Syntax: head(Vars) :- body_atom1(Vars), body_atom2(Vars).
# Variables start with uppercase; trailing '.' is optional
dr.add_rule("ancestor(X, Y) :- parent(X, Y).")
dr.add_rule("ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).")  # recursive

print(f"Rules loaded: {len(dr._rules)}")

In [ ]:

# ── Step 4: fixpoint evaluation ────────────────────────────────────────────
# derive_all() runs semi-naive bottom-up evaluation until no new facts appear
all_facts: list[str] = dr.derive_all()

ancestor_strs = sorted(f for f in all_facts if f.startswith("ancestor"))
print(f"Derived {len(ancestor_strs)} ancestor facts:")
for f in ancestor_strs:
    print(" ", f)

In [ ]:

# ── Step 5: query ──────────────────────────────────────────────────────────
# Use '?varname' placeholders — query() auto-calls derive_all() if needed
# Returns: list[dict]  e.g. [{"Y": "bob"}, {"Y": "ann"}, {"Y": "pat"}]

descendants = dr.query("ancestor(tom, ?Y)")
print("All descendants of tom:", sorted(r["Y"] for r in descendants))

ancestors_of_pat = dr.query("ancestor(?X, pat)")
print("All ancestors of pat: ", sorted(r["X"] for r in ancestors_of_pat))

all_pairs = dr.query("ancestor(?X, ?Y)")
print(f"\nAll ancestor pairs ({len(all_pairs)}):")
for row in sorted(all_pairs, key=lambda r: (r["X"], r["Y"])):
    print(f"  {row['X']:6s} → {row['Y']}")

---

Part 2 — GraphBuilder → DatalogReasoner Pipeline

GraphBuilder constructs a structured {"entities": [...], "relationships": [...]} dict from your data. We then:

1. Analyse the graph with GraphAnalyzer to understand structure.
2. Feed kg["relationships"] into DatalogReasoner as EDB facts.
3. Apply recursive Datalog rules over the KG.

In [ ]:

# ── Build a software-dependency KG ────────────────────────────────────────
entities = [
    {"id": "pythonsdk",   "name": "Python SDK",   "type": "Component"},
    {"id": "restapi",     "name": "REST API",     "type": "Component"},
    {"id": "authservice", "name": "Auth Service", "type": "Component"},
    {"id": "database",    "name": "Database",     "type": "Component"},
    {"id": "dashboard",   "name": "Dashboard",    "type": "Component"},
    {"id": "analytics",   "name": "Analytics",    "type": "Component"},
]
relationships = [
    {"source": "pythonsdk",   "target": "restapi",     "type": "depends_on"},
    {"source": "restapi",     "target": "authservice", "type": "depends_on"},
    {"source": "authservice", "target": "database",    "type": "depends_on"},
    {"source": "dashboard",   "target": "restapi",     "type": "depends_on"},
    {"source": "dashboard",   "target": "analytics",   "type": "depends_on"},
    {"source": "analytics",   "target": "database",    "type": "depends_on"},
]

# GraphBuilder validates, deduplicates, and packages the data
builder = GraphBuilder(merge_entities=True, resolve_conflicts=False)
kg = builder.build([{"entities": entities, "relationships": relationships}])

print(f"KG built — entities: {len(kg['entities'])}, relationships: {len(kg['relationships'])}")

In [ ]:

# ── Analyse the graph structure before reasoning ───────────────────────────
# GraphAnalyzer provides centrality, communities, connectivity, and metrics
analyzer = GraphAnalyzer()
metrics = analyzer.compute_metrics(graph=kg)

print("Graph structure:")
print(f"  Nodes      : {metrics['num_nodes']}")
print(f"  Edges      : {metrics['num_edges']}")
if "density" in metrics:
    print(f"  Density    : {metrics['density']:.3f}")
if "is_connected" in metrics:
    print(f"  Connected  : {metrics['is_connected']}")

In [ ]:

# ── Load KG relationships as EDB facts ────────────────────────────────────
# GraphBuilder output dicts use the same source/target/type shape that
# DatalogReasoner.add_fact() natively understands
dr = DatalogReasoner()

for rel in kg["relationships"]:
    dr.add_fact(rel)   # dict path: {"source": ..., "target": ..., "type": ...}

print(f"EDB loaded: {len(dr._all_facts)} dependency facts")

In [ ]:

# ── Transitive dependency closure ─────────────────────────────────────────
# 'depends_on' is the predicate name that add_fact inferred from 'type'
dr.add_rule("transitive_dep(X, Y) :- depends_on(X, Y).")
dr.add_rule("transitive_dep(X, Y) :- depends_on(X, Z), transitive_dep(Z, Y).")

dr.derive_all()

# Everything that transitively depends on the database
db_deps = sorted(r["X"] for r in dr.query("transitive_dep(?X, database)"))
print("Components that transitively depend on Database:")
for c in db_deps:
    print(" ", c)

# What does pythonsdk transitively depend on?
sdk_chain = sorted(r["Y"] for r in dr.query("transitive_dep(pythonsdk, ?Y)"))
print(f"\nPython SDK full dependency chain: {sdk_chain}")

---

Part 3 — ContextGraph + load_from_graph()

DatalogReasoner.load_from_graph(graph) accepts any ContextGraph directly: it calls graph.find_edges() and graph.find_nodes() and converts each result into EDB facts automatically.

In [ ]:

# ── Build an in-memory ContextGraph ───────────────────────────────────────
# ContextGraph.add_node / add_edge are the canonical way to build in-memory KGs
cg = ContextGraph()

# Nodes
for person in ["alice", "bob", "carol", "dave", "eve"]:
    cg.add_node(person, node_type="person", name=person.capitalize())

# Directed "follows" edges
for src, dst in [("alice", "bob"), ("bob", "carol"), ("carol", "dave"), ("alice", "eve"), ("eve", "carol")]:
    cg.add_edge(src, dst, edge_type="follows")

# Verify the graph built correctly
nodes = cg.find_nodes(node_type="person")
edges = cg.find_edges(edge_type="follows")
print(f"ContextGraph — nodes: {len(nodes)}, edges: {len(edges)}")
print("Edges:", [(e.get("source", e.get("source_id")), e.get("target", e.get("target_id"))) for e in edges])

In [ ]:

# ── load_from_graph() ingests the ContextGraph directly ───────────────────
dr = DatalogReasoner()
n_loaded = dr.load_from_graph(cg)  # calls cg.find_edges() + cg.find_nodes() internally
print(f"Facts loaded from ContextGraph: {n_loaded}")

In [ ]:

# ── Influence reach via transitive 'follows' ──────────────────────────────
dr.add_rule("influence(X, Y) :- follows(X, Y).")
dr.add_rule("influence(X, Y) :- follows(X, Z), influence(Z, Y).")

dr.derive_all()

# Who can alice reach?
alice_reach = sorted(r["Y"] for r in dr.query("influence(alice, ?Y)"))
print(f"Alice's influence reach : {alice_reach}")

# Who can reach dave?
reach_dave = sorted(r["X"] for r in dr.query("influence(?X, dave)"))
print(f"Who can influence dave  : {reach_dave}")

# Full influence matrix
all_influence = dr.query("influence(?X, ?Y)")
print(f"\nTotal influence pairs: {len(all_influence)}")

---

Part 4 — RBAC Access-Control Policy

We model a role-based access-control (RBAC) system:

1. Use GraphBuilder to build a structured KG of users, roles, and permissions.
2. Load it into DatalogReasoner for policy inference.
3. Use ExplanationGenerator to produce audit-ready NL justifications.

In [ ]:

# ── Build RBAC graph with GraphBuilder ────────────────────────────────────
rbac_entities = [
    # Users
    {"id": "alice",  "type": "User", "name": "Alice"},
    {"id": "bob",    "type": "User", "name": "Bob"},
    {"id": "carol",  "type": "User", "name": "Carol"},
    {"id": "dave",   "type": "User", "name": "Dave"},
    # Roles
    {"id": "admin",  "type": "Role", "name": "Administrator"},
    {"id": "editor", "type": "Role", "name": "Editor"},
    {"id": "viewer", "type": "Role", "name": "Viewer"},
    # Permissions
    {"id": "read",         "type": "Permission"},
    {"id": "write",        "type": "Permission"},
    {"id": "delete",       "type": "Permission"},
    {"id": "manage_users", "type": "Permission"},
]
rbac_relationships = [
    # User → Role assignments
    {"source": "alice",  "target": "admin",  "type": "has_role"},
    {"source": "bob",    "target": "editor", "type": "has_role"},
    {"source": "carol",  "target": "viewer", "type": "has_role"},
    {"source": "dave",   "target": "editor", "type": "has_role"},
    # Role hierarchy (admin inherits from editor, editor from viewer)
    {"source": "admin",  "target": "editor", "type": "role_inherits"},
    {"source": "editor", "target": "viewer", "type": "role_inherits"},
    # Role → Permission grants
    {"source": "viewer", "target": "read",         "type": "role_has_perm"},
    {"source": "editor", "target": "write",        "type": "role_has_perm"},
    {"source": "admin",  "target": "delete",       "type": "role_has_perm"},
    {"source": "admin",  "target": "manage_users", "type": "role_has_perm"},
]

builder = GraphBuilder(merge_entities=True, resolve_conflicts=False)
rbac_kg = builder.build([{"entities": rbac_entities, "relationships": rbac_relationships}])

print(f"RBAC KG — entities: {len(rbac_kg['entities'])}, relationships: {len(rbac_kg['relationships'])}")

In [ ]:

# ── Analyse RBAC graph structure ──────────────────────────────────────────
analyzer = GraphAnalyzer()
metrics = analyzer.compute_metrics(graph=rbac_kg)
centrality = analyzer.calculate_centrality(rbac_kg, centrality_type="degree")

print(f"RBAC graph — {metrics['num_nodes']} nodes, {metrics['num_edges']} edges")
if isinstance(centrality, dict) and "degree" in centrality:
    top = sorted(centrality["degree"].items(), key=lambda x: x[1], reverse=True)[:3]
    print("Top-3 nodes by degree centrality:", top)

In [ ]:

# ── Load RBAC KG into DatalogReasoner ────────────────────────────────────
dr = DatalogReasoner()

for rel in rbac_kg["relationships"]:
    dr.add_fact(rel)   # {source, target, type} → predicate(source, target)

# ── IDB rules: transitive role hierarchy ─────────────────────────────────
dr.add_rule("effective_role(R, R2) :- role_inherits(R, R2).")
dr.add_rule("effective_role(R, R2) :- role_inherits(R, Z), effective_role(Z, R2).")

# ── IDB rules: inherited permissions ─────────────────────────────────────
dr.add_rule("role_can(R, P) :- role_has_perm(R, P).")
dr.add_rule("role_can(R, P) :- effective_role(R, R2), role_has_perm(R2, P).")

# ── IDB rules: user effective permissions ────────────────────────────────
dr.add_rule("can(U, P) :- has_role(U, R), role_can(R, P).")

dr.derive_all()

print("User permissions derived via role-hierarchy inference:")
for user in ["alice", "bob", "carol", "dave"]:
    perms = sorted(r["P"] for r in dr.query(f"can({user}, ?P)"))
    print(f"  {user:6s}: {perms}")

In [ ]:

# ── ExplanationGenerator — audit-ready NL justification ──────────────────
# ExplanationGenerator works with InferenceResult objects.
# We construct one manually to represent a derived Datalog conclusion.

explainer = ExplanationGenerator(detail_level="detailed")

# Build the Rule object that represents the permission derivation chain
perm_rule = Rule(
    rule_id="rbac_perm_chain",
    name="RBAC permission via role hierarchy",
    conditions=["has_role(alice, admin)", "effective_role(admin, viewer)", "role_has_perm(viewer, read)"],
    conclusion="can(alice, read)",
    rule_type=RuleType.IMPLICATION,
    confidence=1.0,
)

# Build InferenceResult representing the Datalog conclusion
result = InferenceResult(
    conclusion="can(alice, read)",
    rule_used=perm_rule,
    premises=[
        "has_role(alice, admin)",
        "role_inherits(admin, editor)",
        "role_inherits(editor, viewer)",
        "role_has_perm(viewer, read)",
    ],
    confidence=1.0,
)

# Generate NL explanation
explanation = explainer.generate_explanation(result)
print("Explanation type :", explanation.explanation_type)
print("Conclusion       :", explanation.conclusion)
print("Natural language :", explanation.natural_language)
print("Reasoning steps  :", len(explanation.reasoning_path.steps))

In [ ]:

# ── Inverse queries ───────────────────────────────────────────────────────
deleters = sorted(r["U"] for r in dr.query("can(?U, delete)"))
print("Who can delete:", deleters)

writers = sorted(r["U"] for r in dr.query("can(?U, write)"))
print("Who can write: ", writers)

# All (user, permission) pairs — full policy matrix
all_caps = dr.query("can(?U, ?P)")
print(f"\nTotal (user, permission) pairs: {len(all_caps)}")

---

Part 5 — Organisation Hierarchy with ContextGraph

We model a company org-chart using ContextGraph and derive:

• manages(M, E) — direct and transitive management
• skip_level(M, E) — two hops up the chain
• same_team(X, Y) — shared team membership

In [ ]:

# ── ContextGraph: org chart ────────────────────────────────────────────────
org = ContextGraph()

# Add employees as nodes with metadata
staff = [
    ("eng1",      "engineer",  "backend"),
    ("eng2",      "engineer",  "backend"),
    ("eng3",      "engineer",  "frontend"),
    ("techlead",  "lead",      "engineering"),
    ("design1",   "designer",  "ux"),
    ("design2",   "designer",  "ux"),
    ("designlead","lead",      "design"),
    ("vpeng",     "vp",        "engineering"),
    ("cto",       "executive", "leadership"),
]
for emp_id, role, team in staff:
    org.add_node(emp_id, node_type="employee", role=role, team=team)

# Reporting lines
reports_to = [
    ("eng1", "techlead"), ("eng2", "techlead"), ("eng3", "techlead"),
    ("techlead", "vpeng"),
    ("design1", "designlead"), ("design2", "designlead"),
    ("designlead", "vpeng"),
    ("vpeng", "cto"),
]
for employee, manager in reports_to:
    org.add_edge(employee, manager, edge_type="reports_to")

# Team membership edges
teams = [
    ("eng1", "backend"),  ("eng2", "backend"),  ("eng3", "frontend"),
    ("design1", "ux"),    ("design2", "ux"),
]
for emp, team in teams:
    org.add_edge(emp, team, edge_type="in_team")
    if not org.find_nodes(node_type="team"):
        org.add_node(team, node_type="team")

print(f"ContextGraph — nodes: {len(org.find_nodes())}, edges: {len(org.find_edges())}")

In [ ]:

# ── Load org chart into DatalogReasoner ───────────────────────────────────
dr = DatalogReasoner()
n = dr.load_from_graph(org)   # uses org.find_edges() + org.find_nodes()
print(f"Facts loaded via load_from_graph(): {n}")

# ── IDB rules ─────────────────────────────────────────────────────────────
# Transitive management chain
dr.add_rule("manages(M, E) :- reports_to(E, M).")
dr.add_rule("manages(M, E) :- reports_to(E, Z), manages(M, Z).")

# Skip-level: exactly two reporting hops
dr.add_rule("skip_level(M, E) :- reports_to(E, Z), reports_to(Z, M).")

# Same team
dr.add_rule("same_team(X, Y) :- in_team(X, T), in_team(Y, T).")

dr.derive_all()

In [ ]:

# ── Query org hierarchy ────────────────────────────────────────────────────
# Everyone under CTO
under_cto = sorted(r["E"] for r in dr.query("manages(cto, ?E)"))
print(f"CTO manages ({len(under_cto)} people): {under_cto}")

# VP Eng's direct + indirect reports
under_vp = sorted(r["E"] for r in dr.query("manages(vpeng, ?E)"))
print(f"VP Eng manages           : {under_vp}")

# Skip-level reports to CTO (people two hops below CTO)
skip = sorted(r["E"] for r in dr.query("skip_level(cto, ?E)"))
print(f"CTO skip-level reports   : {skip}")

# eng1's teammates
mates = [r["Y"] for r in dr.query("same_team(eng1, ?Y)") if r["Y"] != "eng1"]
print(f"eng1's teammates         : {sorted(mates)}")

In [ ]:

# ── Build an InferenceResult and explain an org query ────────────────────
explainer = ExplanationGenerator(detail_level="verbose")

mgmt_rule = Rule(
    rule_id="transitive_manages",
    name="Transitive management chain",
    conditions=["reports_to(eng1, techlead)", "manages(vpeng, techlead)"],
    conclusion="manages(vpeng, eng1)",
    rule_type=RuleType.IMPLICATION,
    confidence=1.0,
)
result = InferenceResult(
    conclusion="manages(vpeng, eng1)",
    rule_used=mgmt_rule,
    premises=["reports_to(eng1, techlead)", "reports_to(techlead, vpeng)"],
    confidence=1.0,
)

exp = explainer.generate_explanation(result)
print(exp.natural_language)

---

Part 6 — Engine Introspection: DatalogFact & DatalogRule

After reasoning, the engine's internal state is fully accessible via DatalogFact and DatalogRule data-classes. Use this for auditing, debugging, or downstream export.

In [ ]:

# ── Inspect DatalogRule objects ────────────────────────────────────────────
# dr._rules  →  List[DatalogRule]
# DatalogRule.head_predicate, .head_args, .body  (body = List[BodyAtom])
print("Rules in engine:")
for rule in dr._rules:
    body_str = ", ".join(
        f"{atom.predicate}({', '.join(atom.args)})"
        for atom in rule.body
    )
    head_str = f"{rule.head_predicate}({', '.join(rule.head_args)})"
    print(f"  {head_str}  :-  {body_str}")

In [ ]:

# ── Inspect DatalogFact objects ────────────────────────────────────────────
# dr._all_facts  →  Set[DatalogFact]  (EDB + IDB combined after derive_all)
# dr._fact_index →  Dict[predicate, Set[DatalogFact]]

from collections import Counter

# Count facts per predicate
predicate_counts = Counter(f.predicate for f in dr._all_facts)
print("Facts per predicate (EDB + derived IDB):")
for pred, count in sorted(predicate_counts.items()):
    print(f"  {pred:20s}: {count}")
print(f"\n  TOTAL: {len(dr._all_facts)}")

In [ ]:

# ── Separate EDB from IDB ─────────────────────────────────────────────────
# EDB predicates are the ones we added via add_fact (not derived by rules)
idb_predicates = {rule.head_predicate for rule in dr._rules}
edb_predicates = {f.predicate for f in dr._all_facts} - idb_predicates

print(f"EDB predicates (base facts)  : {sorted(edb_predicates)}")
print(f"IDB predicates (derived)     : {sorted(idb_predicates)}")

In [ ]:

# ── Sample DatalogFact structure ──────────────────────────────────────────
# DatalogFact is a frozen dataclass: predicate: str, args: Tuple[str, ...]
manages_facts = sorted(dr._fact_index.get("manages", []), key=lambda f: f.args)
print(f"First 5 'manages' DatalogFact objects ({len(manages_facts)} total):")
for fact in manages_facts[:5]:
    # Access predicate and args directly from the dataclass
    print(f"  DatalogFact(predicate={fact.predicate!r}, args={fact.args})")

In [ ]:

# ── clear() resets the engine completely ─────────────────────────────────
print(f"Facts before clear(): {len(dr._all_facts)}")
dr.clear()
print(f"Facts after  clear(): {len(dr._all_facts)}")
print(f"Rules after  clear(): {len(dr._rules)}")

---

API Summary

DatalogReasoner

Method	Input	Output	Notes
add_fact(f)	str or dict	None	string: "pred(a, b)" · dict: {source, target, type}
add_rule(s)	str	None	Horn clause: "head(X) :- body(X, Y)."
derive_all()	—	list[str]	semi-naive fixpoint; idempotent
query(pat)	str	list[dict]	"pred(a, ?Y)" → [{"Y": ...}]
load_from_graph(g)	ContextGraph	int	facts loaded count
clear()	—	None	resets engine

Syntax rules

Item	Rule	Example
Variable	Starts uppercase	X, Role, Parent
Constant	All lowercase	tom, admin, database
Query var	Prefix ?	?X, ?Y, ?Role
Rule body	:- separator, comma between atoms	head(X) :- a(X, Z), b(Z, Y).

Class map

GraphBuilder.build()            → kg dict {entities, relationships}
                   ↓ kg["relationships"] → dr.add_fact(rel)
                   
ContextGraph.add_node/add_edge  → in-memory graph
                   ↓ dr.load_from_graph(cg)
                   
DatalogReasoner.add_rule()      → Horn clause rules
DatalogReasoner.derive_all()    → semi-naive fixpoint
DatalogReasoner.query()         → result rows
                   ↓ build InferenceResult
                   
ExplanationGenerator            → natural language justification
GraphAnalyzer                   → graph structure metrics pre/post reasoning
DatalogFact / DatalogRule       → introspect engine state