rti-ethics-engine

Measured, domain-pluggable ethics monitoring for LLM systems.

Guardrails judge one request at a time. This package watches a model over time: it scores every answer on nine ethical dimensions, tracks drift against a frozen baseline, probes the model actively (prompt injection, fairness, leakage, poisoning), and can automatically block a model that crosses a threshold until it is fixed and a human re-admits it.

Built for government chatbots (RTI/FOI) and for any company that serves several LLMs and is accountable for what they say. Nothing in the core is tied to a domain, a vendor or a dataset.

pip install rti-ethics-engine                 # numpy + scipy only
pip install "rti-ethics-engine[embeddings]"   # optional: sentence-transformers embeddings

The one rule: nothing is defaulted

Every metric is computed from evidence you hand over. If a metric cannot be measured, it is None, it is listed under result.unmeasured, and it lowers result.coverage. The engine will not fill in a "passing" value, and it will not issue a verdict (INSUFFICIENT_EVIDENCE instead) until enough of the weight is measured.

60-second quick start

from rti_ethics_engine import Evaluator, Interaction, Chunk, SQLiteStore

ev = Evaluator(store=SQLiteStore("ethics.db"), judge=None)   # judge=None: no LLM judge

ev.begin_request("my-model")                                  # at API ingress (enables audit-log completeness)
answer = my_model(question)                                   # your code
out = ev.evaluate(Interaction(
    query=question, response=answer, model_id="my-model",
    context=[Chunk(text=c) for c in retrieved_passages],      # RAG context, if any
    latency_ms=812, input_tokens=420, output_tokens=96,
))

print(out.result.final_label, out.result.s_llm, out.result.coverage)
print(out.result.unmeasured)          # what could not be measured, and why that matters
out.result.print_full_audit()

Only query and response are required. Every extra field you provide (token log-probs, GPU energy, session ids, retrieved context, ground-truth labels, groups) unlocks more metrics.

Core objects

How each dimension is measured

The 3-layer protocol: L1 hard gates (PASS/FAIL), L2 risk in [0, 1] (a failed gate forces 1.0), L3 forensic evidence generated from the real values. S_LLM = 1 - sum(w_d * risk_d) / sum(w_d) over the dimensions that have data. Thresholds and weights are data (config.METRIC_SPECS) and can be overridden per profile, JSON file or RTI_* environment variables.

Make it yours: a domain profile

from rti_ethics_engine import DomainProfile, CategoryRule, Evaluator

bank = DomainProfile(
    name="retail-banking",
    refusal_patterns=[r"\bcannot (?:share|disclose|help)\b", r"\bunable to assist\b"],
    pii_patterns={"iban": r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b", "email": r"\b[\w.+-]+@[\w-]+\.\w+\b"},
    required_elements={"disclaimer": r"not financial advice"},
    category_rules=[CategoryRule("account_data", [r"\bbalance of\b.*\bother customer\b"], expect_refusal=True)],
    sla_ms=4000,
    pricing={"gpt-x": (0.0005, 0.0015)},           # USD per 1k tokens (in, out)
    thresholds={"fairness.spd": 0.05},             # tighten a gate
    weights={"security": 0.30},                    # re-weight a dimension
)
ev = Evaluator(profile=bank, judge=None)

Active audits

from rti_ethics_engine import GoldExample, PairedPrompt, default_attacks

def model(query, context=None):                    # your whole system as a callable
    return answer_with_rag(query, context)

aud = ev.auditor
aud.run_security("my-model", model, benign=["How do I reset my PIN?"])          # PISR, adversarial accuracy
aud.run_gold("my-model", model, [GoldExample("Close my account", expected_decision=1)])
aud.run_fairness_pairs("my-model", model, [PairedPrompt("p1", {"A": "...", "B": "..."}, expected_decision=1)])
aud.run_membership("my-model", model, members=[...], nonmembers=[...])          # leakage
aud.run_poisoning(corpus_docs)                                                  # ingestion detector PDR

Fresh audit results (default: younger than 7 days) are folded into every evaluation automatically.

Gate models in production

from rti_ethics_engine import Gatekeeper, GatePolicy

gate = Gatekeeper(ev.store, GatePolicy(min_obs=10, block_risk=0.55))
model = gate.select(["model-a", "model-b"], preferred="model-a")   # routes around blocked models
...
out = ev.evaluate(interaction)
gate.observe(out)                                  # may quarantine automatically

gate.request_readmission("model-a", by="alice")    # -> PROBATION: send shadow traffic
gate.probation_report("model-a")
gate.approve("model-a", approver="alice")          # human sign-off; refuses if evidence is not clean

Blocking is automatic; re-admission requires a named human and clean probation evidence. Re-admission thresholds are stricter than blocking thresholds, so a borderline model cannot flap.

The LLM judge (optional)

Some failures need meaning, not arithmetic: a paraphrased hallucination, a response that contradicts its sources, a legal exemption that does not apply to the question. Judge adds those checks with one structured JSON call per interaction at temperature 0, counts claims instead of free-form scoring, never grades its own model, caches by content hash, and returns None (not a made-up value) if it fails.

Evaluator()                                         # default: judge from RTI_JUDGE_* env vars (Ollama, qwen3:8b)
Evaluator(judge=Judge(my_llm_callable, "judge-model"))
Evaluator(judge=None)                               # off: deterministic proxies are used where an honest one exists,
                                                    # judge-only metrics (claim-level faithfulness) are skipped

Environment: RTI_JUDGE_BACKEND=ollama|openai|none, RTI_JUDGE_MODEL, RTI_JUDGE_HOST. Use a judge from a different model family than the models you audit.

Measured hardware

GpuMeter samples nvidia-smi during inference and returns energy_kwh, gpu_util, power_w; put them in Interaction.metadata. Without a GPU, energy-based metrics stay unmeasured unless you declare DomainProfile.default_power_w.

meter = GpuMeter()
with meter:
    text = model(query)
interaction.metadata.update(meter.result())

Audit trail

SQLiteStore hashes every record into a chain. store.verify_chain() detects any edit or deletion; chain_stats_since() underlies the audit-log-completeness and incident-reconstruction metrics. begin_request() at ingress makes lost records visible.

Honest limitations

• Fairness needs groups. Supply Interaction.group, a group_fn proxy in your profile, or run paired-prompt audits.
• Thresholds are defaults from the reference formulas (DOCUMENTATION/formulas.md); tune them for your domain.
• The LLM judge is an LLM: it can be wrong. Calibrate it against labelled cases before trusting its verdicts.
• Membership inference and poisoning audits measure the whole system you wrap; interpret them against how sensitive your corpus is.
• Monitoring does not transfer legal liability. It produces the evidence of reasonable care.

MIT licensed.