Skip to content

Python API reference

Everything below is importable from hypermesh.rag.symbolic (symbolic core) or hypermesh.rag (pipeline + generation).

from hypermesh.rag.symbolic import (
parse_rule, Rule, Atom, Compare, Neg, Head, RuleValidationError, EDB_PREDS, is_var,
Fact, FactBase, FactMeta, facts_from_edges,
SymbolicReasoner, ReasonResult, StratificationError, stratify, match_body,
build_proof, ProofTree,
HallucinationFirewall, FirewallResult,
RuleStore,
)
from hypermesh.rag import RAGPipeline, RAGResult, LLMConfig, GenerationResult

Parse and validate a rule dict. Raises RuleValidationError if malformed or unsafe (not range-restricted). Accepts if/then or body/head keys.

True if x is a ?-prefixed variable string matching ^\?[A-Za-z_][A-Za-z0-9_]*$.

The reserved base predicates: {"hyperedge", "formation", "member", "member_type", "weight", "at"}. A rule head may not redefine these.

MemberTypeMeaning
idstrUnique id.
headHeadConclusion atom.
bodylist[Atom|Compare|Neg]Body literals.
name, version, priority, enabledstr/int/int/boolMetadata.
provenancedictFree-form.
positives (property)list[Atom]Positive body atoms.
compares (property)list[Compare]Comparison literals.
negations (property)list[Neg]Negated literals.
body_pos_preds / body_neg_preds (property)set[str]Predicate names.
as_dict()dictJSON round-trip form.

A positive literal. Frozen/hashable.

Arithmetic/relational guard. op ∈ {>=, <=, >, <, ==, !=}.

Negation-as-failure over a positive atom.

class Headpred: str, args: tuple, confidence: float = 1.0

Section titled “class Head — pred: str, args: tuple, confidence: float = 1.0”

The rule conclusion.

Raised for malformed or unsafe rules.

facts_from_edges(edges, table="") -> FactBase

Section titled “facts_from_edges(edges, table="") -> FactBase”

Project retrieved hyperedges (RetrievedEdge) into a base FactBase, emitting hyperedge/formation/weight/at/member/member_type facts with edge_ref provenance per edge.

A ground atom. Frozen/hashable; str(fact)pred(a, b, …).

source ("base"/"derived"), confidence, support: tuple[Fact, ...], rule_id, edge_ref.

MethodReturnsMeaning
add_base(fact, *, confidence=1.0, edge_ref=None)boolAdd base fact; True if new.
add_derived(fact, support, rule_id, confidence)boolAdd derived fact; True if new.
by_pred(pred)list[Fact]Facts for a predicate (indexed).
meta(fact)FactMetaMetadata for a fact.
derived()list[Fact]All derived facts.
__contains__ / __len__ / __iter__Membership / size / iterate.

Reasoner — hypermesh.rag.symbolic.reasoner

Section titled “Reasoner — hypermesh.rag.symbolic.reasoner”

Assign each predicate a stratum (EDB = 0; negated predicate strictly below the head that negates it). Raises StratificationError on negation through recursion.

match_body(rule, fb) -> Iterator[tuple[dict, list[Fact]]]

Section titled “match_body(rule, fb) -> Iterator[tuple[dict, list[Fact]]]”

Yield (binding, support_facts) for every way rule’s body is satisfied over fb (positive join → comparisons → negation-as-failure).

SymbolicReasoner(max_iterations=64, deadline_ms=None). run(fact_base, rules) -> ReasonResult — stratified forward chaining to a fixpoint, writing derived facts back into fact_base.

fact_base, derived: list[Fact], rules_fired: list[str], iterations: int, strata: dict[str, int], timed_out: bool.

Raised when a rule set cannot be stratified.

build_proof(goal, fb, *, proof_id="proof", provenance_resolver=None) -> ProofTree

Section titled “build_proof(goal, fb, *, proof_id="proof", provenance_resolver=None) -> ProofTree”

Build a proof DAG for goal from a reasoned FactBase. Returns an unproved tree if the goal isn’t present.

proof_id, goal, status ("proved"/"unproved"), confidence, depth, rules_fired, nodes, edges. as_dict() → JSON; evidence_tags (property) → set[str] of all HEDGE-/RULE-/STEP- tags.

Type alias: Callable[[dict], dict]. Receives a leaf’s edge_ref, returns extra provenance fields merged into the leaf.

Firewall — hypermesh.rag.symbolic.firewall

Section titled “Firewall — hypermesh.rag.symbolic.firewall”

HallucinationFirewall(abstain_message=...). check(answer, allowed_tags, *, proved=True, require_proof=False) -> FirewallResult — strip factual sentences that don’t cite an allowed tag; abstain when required proof is missing or nothing is left.

answer, outcome (supported/stripped/abstained), abstained, coverage, supported: list[str], stripped: list[str]. as_dict(){outcome, abstained, coverage, supported_count, stripped_count, stripped}.

RuleStore(db_dir="data") — JSON registry at {db_dir}/rag_rules.json.

MethodReturns
upsert(rule_dict)rule dict + stratum (validates + stratifies whole set)
validate(rule_dict){ok, errors, stratum?} (no write)
get(rule_id)dict | None
list()list of summaries
delete(rule_id)bool
all_rules()list[Rule]
enabled_rules()list[Rule]
path (property)str

See Pipeline, modes & LLMs for the full constructor table.

MemberSignatureMeaning
__init__(db, table, …, mode="neuro", require_proof=False, rules=None, rule_store=None, …)Configure the pipeline.
queryasync (query_text) -> RAGResultFull flow with firewall.
streamasync (query_text) -> (ParsedQuery, list[RetrievedEdge], AsyncIterator[str])Token streaming (generation only).
table / modelpropertyTable name / model name.

Full field list in Pipeline → RAGResult. as_dict() returns the JSON form used by the REST API.

Backends and fields in Pipeline → Plug in any LLM. Class methods: mock(), custom(chat_fn, model="custom", **kwargs), ollama(model="phi3:mini"), lmstudio(model="local-model"), llama_cpp(gguf_path, n_gpu_layers=-1).

answer, cited_edge_ids, confidence, model, prompt_tokens, completion_tokens, elapsed_ms, low_confidence_warning. as_dict() → JSON.

ExceptionRaised byWhen
RuleValidationErrorparse_rule, RuleStore.upsertMalformed/unsafe rule.
StratificationErrorstratify, SymbolicReasoner.runNegation through recursion.
ValueErrorRAGPipelineInvalid mode, table not found.
TypeErrorLLMConfig.customchat_fn not callable / bad return type.