Case Study: Verifiable RAG for AML (end-to-end)

This walkthrough takes a small anti-money-laundering (AML) scenario from raw data to a provable, audit-ready answer, exercising the whole verifiable RAG pipeline in a single script:

retrieve → reason (rules) → prove → generate (LLM) → firewall

The complete, self-contained script is reproduced in full at the end of this page — copy it into a file and run it. It needs no API key, no GPU, and no network — it uses the deterministic mock LLM so the output below is exactly reproducible.

pip install hypermesh
python case_study_neurosymbolic_rag.py

See the Neuro-Symbolic RAG section for the concepts, the rule schema, and the Python API reference.

1. The graph

Transactions are a textbook hypergraph: one transaction connects several accounts at once. We model six accounts (two are known shell companies) and five timestamped transactions; each transaction’s weight is a risk score.

import hypermeshdb

ACCOUNTS = {
    1001: {"short": "Acme-LLC",        "type": "account"},
    1002: {"short": "Bridge-Holdings", "type": "shell"},   # known shell company
    1003: {"short": "Personal-A",      "type": "account"},
    1004: {"short": "Offshore-X",      "type": "shell"},   # known shell company
    1005: {"short": "Retail-Z",        "type": "account"},
    1006: {"short": "Personal-B",      "type": "account"},
}

db = hypermeshdb.connect(db_dir)
db.execute("CREATE HYPEREDGE TABLE TRANSACTIONS () BUCKET_SECONDS 1")
db.insert(event_ts=1000, members=[1001, 1002],       weight=0.91, table="TRANSACTIONS")
db.insert(event_ts=1000, members=[1003, 1005],       weight=0.30, table="TRANSACTIONS")
db.insert(event_ts=2000, members=[1001, 1004, 1002], weight=0.86, table="TRANSACTIONS")
db.insert(event_ts=2500, members=[1005, 1006],       weight=0.40, table="TRANSACTIONS")
db.insert(event_ts=3000, members=[1003, 1004],       weight=0.82, table="TRANSACTIONS")

2. Policy as rules

Compliance policy is written as declarative, auditable rules — not buried in prompt text. This set shows all three reasoning features: a base rule, a multi-step escalation, and stratified negation.

RULES = [
    {   # high-risk transaction touching a shell company → structuring
        "id": "rule:structuring",
        "if": [
            {"pred": "member",      "args": ["?E", "?A"]},
            {"pred": "member_type", "args": ["?A", "shell"]},
            {"pred": "weight",      "args": ["?E", "?W"]},
            {"compare": ["?W", ">=", 0.8]},
        ],
        "then": {"pred": "structuring", "args": ["?E"], "confidence": 0.9},
    },
    {   # multi-step: structuring escalates to a SAR candidate
        "id": "rule:sar_candidate",
        "if": [{"pred": "structuring", "args": ["?E"]}],
        "then": {"pred": "sar_candidate", "args": ["?E"], "confidence": 0.95},
    },
    {   # stratified negation: not-structuring → routine
        "id": "rule:routine",
        "if": [
            {"pred": "hyperedge", "args": ["?E"]},
            {"not": {"pred": "structuring", "args": ["?E"]}},
        ],
        "then": {"pred": "routine", "args": ["?E"], "confidence": 1.0},
    },
]

3. Reason — LLM-free, with proofs

In symbolic mode the pipeline derives facts and proofs with no LLM at all — ideal for compliance jobs and CI.

from hypermeshdb.rag import RAGPipeline, LLMConfig

pipe = RAGPipeline(db, table="TRANSACTIONS", entity_map=ENTITY_MAP,
                   mode="symbolic", rules=RULES, top_k=50,
                   llm_config=LLMConfig.mock())
res = await pipe.query("Which transactions are SAR candidates?")

[STEP-3] sar_candidate(0)   via rule:sar_candidate  conf=0.85  evidence=['HEDGE-0', 'HEDGE-2']
[STEP-4] sar_candidate(2)   via rule:sar_candidate  conf=0.85  evidence=['HEDGE-2']
[STEP-5] sar_candidate(4)   via rule:sar_candidate  conf=0.85  evidence=['HEDGE-4']
[STEP-6] structuring(0)     via rule:structuring    conf=0.90  evidence=['HEDGE-0', 'HEDGE-2']
[STEP-7] structuring(2)     via rule:structuring    conf=0.90  evidence=['HEDGE-2']
[STEP-8] structuring(4)     via rule:structuring    conf=0.90  evidence=['HEDGE-4']
[STEP-1] routine(1)         via rule:routine        conf=1.00  evidence=['HEDGE-1']
[STEP-2] routine(3)         via rule:routine        conf=1.00  evidence=['HEDGE-3']

Three high-risk shell transactions are flagged as SAR candidates; the two retail transfers are proved routine via negation. Nothing here is a model opinion — it’s a deduction from your data and your policy.

4. Prove — a machine-checked proof

Every derived fact carries a proof DAG whose leaves are real stored hyperedges (verified=True):

status=proved  confidence=0.855  depth=2
  STEP-1 sar_candidate(0)        ⟵ rule rule:sar_candidate [RULE-0]
  STEP-2 structuring(0)          ⟵ rule rule:structuring   [RULE-1]
  HEDGE-0 evidence member(0, 1002)         (verified=True)
  HEDGE-2 evidence member_type(1002, shell)(verified=True)
  HEDGE-0 evidence weight(0, 0.91)         (verified=True)

Read bottom-up: transaction 0 has member 1002, which is a shell, and its risk weight is 0.91 ≥ 0.8 → structuring → SAR candidate. That is the exact, inspectable justification an auditor or regulator can replay.

5. Generate + firewall — a grounded answer

In hybrid mode the proven facts are handed to the LLM, which must cite them; the firewall then strips anything uncited.

pipe = RAGPipeline(db, table="TRANSACTIONS", entity_map=ENTITY_MAP,
                   mode="hybrid", rules=RULES, top_k=50,
                   llm_config=LLMConfig.mock())   # swap for OpenAI / Anthropic / local
res = await pipe.query("Summarize the suspicious transactions.")
print(res.firewall)
# {"outcome": "stripped", "abstained": false, "coverage": 0.947,
#  "supported_count": 18, "stripped_count": 1, ...}

Every claim in the returned answer ends with a citation that traces back to a proof. (Note the firewall even stripped the model’s own uncited preamble.)

6. The firewall blocks a hallucinating model

Swap in a model that fabricates — the guarantee holds anyway:

def lying_model(messages, cfg):
    return ("Bridge-Holdings wired $4.2M to a sanctioned entity in Tehran "
            "on March 3rd, and the CEO personally approved it.")

pipe = RAGPipeline(db, table="TRANSACTIONS", entity_map=ENTITY_MAP,
                   mode="hybrid", rules=RULES, top_k=50,
                   llm_config=LLMConfig.custom(lying_model))
res = await pipe.query("What did Bridge-Holdings do?")

firewall outcome : abstained
returned to user : 'Insufficient verifiable evidence to answer this question.'

The fabricated, uncited claim never reaches the user. With require_proof=True and no firing rules, the pipeline likewise abstains instead of guessing.

7. At scale: the same approach on 5 M real transactions

The toy graph above shows the mechanics. To show it is real and scalable, we run the same neuro-symbolic recipe on the public IBM “AMLworld” HI-Small dataset — 5,078,345 real-style financial transactions with ground-truth laundering labels. This is genuinely store-backed: every train-period transaction is ingested into the HyperMesh engine as a {from_account, to_account} hyperedge, indexed on disk, and its structural signal is read back out of the index — so the claim “ingested as hypergraphs” is literally true.

import hypermeshdb

db = hypermeshdb.connect(db_dir)
db.execute("CREATE HYPEREDGE TABLE TXN () BUCKET_SECONDS 3600")

# 1. Ingest train-period transactions as hyperedges (group-commit WAL).
with db.transaction():
    for ts, src, dst, amount in train_edges:
        db.insert(event_ts=ts, members=[src, dst], weight=amount, table="TXN")

# 2. Build the on-disk TPI + FMI index.
store = db._get_store("TXN")
store.compact()

# 3. Read each account's hypergraph node degree back out of the FMI.
degree = {e["node_id"]: e["appearances"] for e in store.coalition_ranking()}

Engine ingest + index (measured, end-to-end):

Stage	Records	Time	Throughput
Ingest (group-commit WAL)	3.55 M	66.4 s	53,566 rows/s
Index build (compact() → TPI + FMI)	3.55 M	6.8 s	—
Structural features (coalition_ranking())	513 K accounts	1.5 s	one O(N log N) pass

Those engine-derived degrees become the “neuro” features for a gradient-boosted detector, trained twice — with and without the hypergraph signal — and scored on a temporal test split at its true class balance (0.15 % laundering):

Detector	PR-AUC	ROC-AUC	Precision@1 % alerts
Without graph features	0.048	0.931	5.6 %
With hypergraph degrees	0.088	0.965	5.3 %

The engine-derived hypergraph features lift PR-AUC by +83 % (0.048 → 0.088) and ROC-AUC by 3.5 points — at a base rate of 0.15 %, the model’s flagged alerts are ~35× more likely to be laundering than a random transaction.

The symbolic layer then attaches a proof to one real alert, citing facts read straight from the engine — e.g. “account 99165 participates in 115 transactions and its counterparty 98968 in 128” (degrees coalition_ranking() returned) — and the firewall blocks a fabricated “cartel boss confessed in Macau” sentence while preserving the cited, engine-grounded claim.

Honest framing

The hypergraph degrees are a ranking-quality lift (PR-AUC / ROC-AUC), not a silver bullet at a fixed 1 % alert budget on this extreme-imbalance dataset. The point of the section is twofold: (1) the engine ingests and indexes 5 M records in seconds (see Performance and Scaling & Performance), and (2) features derived from the index itself carry real, measurable detection signal — while every surfaced answer stays provable and firewalled.

Why this matters for you

• Audit-ready by construction — every answer ships with a replayable proof down to source records.
• Hallucination is structurally impossible to return — unsupported text is stripped or the system abstains.
• Policy is explicit and versioned — rules live in a durable store, not in prompt strings, and are re-validated for safety on every change.
• Model-agnostic — OpenAI, a local GGUF, or a bring-your-own model all sit behind the same firewall, so you can pick on cost/latency without risking trust.

Full script

The complete, self-contained source for this walkthrough. Save it as case_study_neurosymbolic_rag.py and run it with python case_study_neurosymbolic_rag.py — no API key, GPU, or network required.

"""
case_study_neurosymbolic_rag.py — verifiable RAG end-to-end on an AML scenario.

A single, self-contained, deterministic, network-free tour of HyperMesh's
neuro-symbolic RAG. It models a tiny anti-money-laundering (AML) graph and shows
the full pipeline:

    retrieve -> reason (rules) -> prove -> generate (LLM) -> firewall

1. Build      — a real HyperMesh DB; transactions are timestamped hyperedges
                over the accounts that co-transact; weight = a risk score.
2. Rules      — declarative, auditable policy as Horn rules (with a multi-step
                derivation and stratified negation).
3. Reason     — LLM-free forward chaining -> derived facts (structuring,
                sar_candidate, routine) each with a machine-checked proof.
4. Prove      — proof trees whose leaves bind to real stored hyperedges.
5. Generate   — an LLM answers, citing the proven facts (uses the offline mock
                model so this script needs no API key).
6. Firewall   — a hallucinating model is caught: fabricated claims never reach
                the user; with require_proof the pipeline abstains.

Usage
-----
    pip install hypermesh
    python case_study_neurosymbolic_rag.py

No network, no GPU, no API key required.
"""
from __future__ import annotations

import asyncio
import json
import tempfile

import hypermeshdb
from hypermeshdb.rag import LLMConfig, RAGPipeline

TABLE = "TRANSACTIONS"

# Accounts (nodes). `type` drives member_type(...) facts the rules match on.
ACCOUNTS = {
    1001: {"short": "Acme-LLC",        "type": "account"},
    1002: {"short": "Bridge-Holdings", "type": "shell"},     # known shell company
    1003: {"short": "Personal-A",      "type": "account"},
    1004: {"short": "Offshore-X",      "type": "shell"},     # known shell company
    1005: {"short": "Retail-Z",        "type": "account"},
    1006: {"short": "Personal-B",      "type": "account"},
}
ENTITY_MAP = {str(nid): meta for nid, meta in ACCOUNTS.items()}

# Transactions (hyperedges): (event_ts, members, risk_score)
TRANSACTIONS = [
    (1_000, [1001, 1002],        0.91),   # high risk, touches a shell
    (1_000, [1003, 1005],        0.30),   # routine retail transfer
    (2_000, [1001, 1004, 1002],  0.86),   # high risk, two shells
    (2_500, [1005, 1006],        0.40),   # routine
    (3_000, [1003, 1004],        0.82),   # high risk, touches a shell
]

# Policy as declarative rules.
RULES = [
    {
        "id": "rule:structuring",
        "name": "High-risk transaction touching a shell company is structuring",
        "priority": 10,
        "if": [
            {"pred": "member",      "args": ["?E", "?A"]},
            {"pred": "member_type", "args": ["?A", "shell"]},
            {"pred": "weight",      "args": ["?E", "?W"]},
            {"compare": ["?W", ">=", 0.8]},
        ],
        "then": {"pred": "structuring", "args": ["?E"], "confidence": 0.9},
    },
    {
        # Multi-step: a structuring event escalates to a SAR candidate.
        "id": "rule:sar_candidate",
        "name": "Structuring escalates to a Suspicious Activity Report candidate",
        "if": [{"pred": "structuring", "args": ["?E"]}],
        "then": {"pred": "sar_candidate", "args": ["?E"], "confidence": 0.95},
    },
    {
        # Stratified negation: anything not flagged as structuring is routine.
        "id": "rule:routine",
        "name": "A transaction with no structuring signal is routine",
        "if": [
            {"pred": "hyperedge", "args": ["?E"]},
            {"not": {"pred": "structuring", "args": ["?E"]}},
        ],
        "then": {"pred": "routine", "args": ["?E"], "confidence": 1.0},
    },
]


def _hr(title: str) -> None:
    print("\n" + "=" * 78)
    print(f"  {title}")
    print("=" * 78)


def _build_db(db_dir: str) -> hypermeshdb.Connection:
    db = hypermeshdb.connect(db_dir)
    db.execute(f"CREATE HYPEREDGE TABLE {TABLE} () BUCKET_SECONDS 1")
    for ts, members, risk in TRANSACTIONS:
        db.insert(event_ts=ts, members=members, weight=risk, table=TABLE)
    return db


def _pipe(db, db_dir, *, mode, llm_config, require_proof=False, rules=RULES):
    return RAGPipeline(
        db, table=TABLE, entity_map=ENTITY_MAP, db_dir=db_dir,
        mode=mode, require_proof=require_proof, rules=rules,
        top_k=50, llm_config=llm_config,
    )


async def main() -> None:
    db_dir = tempfile.mkdtemp(prefix="hm_nsrag_")
    db = _build_db(db_dir)

    _hr("1. The graph")
    print(f"  {len(ACCOUNTS)} accounts, {len(TRANSACTIONS)} transactions in table {TABLE!r}.")
    print("  Transactions are hyperedges over the accounts that co-transact; "
          "weight = risk score.")

    # 2. Symbolic mode: reasoning + proofs, no LLM
    _hr("2. Symbolic reasoning (LLM-free) — derived facts + proofs")
    sym = _pipe(db, db_dir, mode="symbolic", llm_config=LLMConfig.mock())
    res = await sym.query("Which transactions are SAR candidates?")
    for f in res.derived_facts:
        print(f"  [{f['step_tag']}] {f['fact']:<22} "
              f"via {f['rule_id']:<22} conf={f['confidence']:.2f} "
              f"evidence={f['hedge_tags']}")

    # Show one full proof tree for a SAR candidate.
    sar = next((p for p in res.proofs if p["goal"]["predicate"] == "sar_candidate"), None)
    if sar:
        _hr("3. A machine-checked proof (sar_candidate)")
        print(f"  status={sar['status']}  confidence={sar['confidence']}  depth={sar['depth']}")
        for n in sar["nodes"]:
            if n["kind"] == "conclusion":
                print(f"    {n['step_tag']} {n['fact']}  <- rule {n['rule']} [{n['rule_tag']}]")
            else:
                v = n["provenance"].get("verified")
                print(f"    {n.get('hedge_tag','')} evidence {n['fact']}  (verified={v})")

    # 4. Hybrid mode: reasoning feeds the LLM, firewall gates the answer
    _hr("4. Hybrid (reason + LLM + firewall) — grounded answer")
    hyb = _pipe(db, db_dir, mode="hybrid", llm_config=LLMConfig.mock())
    res = await hyb.query("Summarize the suspicious transactions.")
    print(res.answer)
    print(f"\n  firewall: {json.dumps(res.firewall)}")

    # 5. The firewall catches a hallucinating model
    _hr("5. Hallucination firewall — a lying model is blocked")

    def lying_model(messages, cfg):
        return ("Bridge-Holdings wired $4.2M to a sanctioned entity in Tehran "
                "on March 3rd, and the CEO personally approved it.")

    liar = _pipe(db, db_dir, mode="hybrid", llm_config=LLMConfig.custom(lying_model))
    res = await liar.query("What did Bridge-Holdings do?")
    print("  model said something fabricated and uncited.")
    print(f"  firewall outcome : {res.firewall['outcome']}")
    print(f"  returned to user : {res.answer!r}")
    print("  -> the fabricated, uncited claim never reaches the user.")

    # 6. require_proof: abstain instead of guessing
    _hr("6. require_proof — abstain when nothing is proved")
    strict = _pipe(db, db_dir, mode="hybrid", llm_config=LLMConfig.mock(),
                   require_proof=True, rules=[])   # no rules -> nothing proved
    res = await strict.query("Is anything suspicious?")
    print(f"  abstained : {res.abstained}")
    print(f"  answer    : {res.answer!r}")

    print("\nDone. Everything above ran offline, deterministically, with no API key.")


if __name__ == "__main__":
    asyncio.run(main())