ai-governance-sdk

ai-governance-sdk is an AI governance SDK for building governed LLM applications with policy enforcement, audit trails, compliance artifacts, approvals, quotas, and observable execution paths.

Install

pip install ai-governance-sdk

Optional extras:

pip install "ai-governance-sdk[google-vertex]"    # Google Vertex AI / Gemini
pip install "ai-governance-sdk[aws-bedrock]"      # Amazon Bedrock
pip install "ai-governance-sdk[azure-openai]"     # Azure OpenAI
pip install "ai-governance-sdk[huggingface]"      # Hugging Face Inference
pip install "ai-governance-sdk[otel]"             # OpenTelemetry tracing
pip install "ai-governance-sdk[postgres]"         # PostgreSQL storage
pip install "ai-governance-sdk[all]"              # Everything

Documentation and API reference: https://api.arelis.digital/docs

End-to-End Developer Demo (Platform)

This SDK supports a full governance lifecycle similar to the TypeScript platform demo:

1. PII scanning and policy gate before model invocation
2. Real model calls (Gemini/Claude) only when gate passes
3. Audit event recording for blocked/allowed runs and tool actions
4. Runtime risk scoring
5. Causal graph construction + commit + lineage
6. Compliance proof generation + verification

Required Environment Variables

export ARELIS_API_KEY="ak_sandbox_..."
export ARELIS_API_URL="https://api.arelis.digital"  # optional, defaults shown below

# only needed if you wire real model calls in your script
export GEMINI_API_KEY="..."
export ANTHROPIC_API_KEY="..."

Quick Start (Copy/Paste)

import os
import re
import uuid
from datetime import datetime, timezone
from typing import Any

from arelis import create_arelis_platform

arelis = create_arelis_platform({
    "baseUrl": os.environ.get("ARELIS_API_URL", "https://api.arelis.digital"),
    "apiKey": os.environ["ARELIS_API_KEY"],
})

PII_PATTERNS = [
    ("email", re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}")),
    ("ssn", re.compile(r"\b\d{3}[-.\s]?\d{2}[-.\s]?\d{4}\b")),
    ("credit_card", re.compile(r"\b(?:\d{4}[-.\s]?){3}\d{4}\b")),
    ("phone", re.compile(r"(?<!\d)(?:\+1[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?)\d{3}[-.\s]?\d{4}(?!\d)")),
]


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def scan_prompt_for_pii(prompt: str) -> dict[str, Any]:
    findings = []
    for name, pattern in PII_PATTERNS:
        for match in pattern.finditer(prompt):
            findings.append({"type": name, "original": match.group()})
    return {"hasPii": len(findings) > 0, "findings": findings}


# 1) Ensure a deny policy exists
policy_key = "pii-deny-before-invocation"
existing = arelis.governance.policies.list({"search": policy_key})
match = next((p for p in existing.get("data", []) if p.get("key") == policy_key), None)
if match:
    policy_id = match["id"]
else:
    created = arelis.governance.policies.create({
        "key": policy_key,
        "name": "PII Deny Before Model Invocation",
        "condition": {"field": "content.pii_detected", "operator": "eq", "value": True},
        "action": "deny",
        "severity": "critical",
        "priority": 1,
    })
    policy_id = created["id"]

# 2) Gate a run
run_id = f"run-{uuid.uuid4()}"
prompt = "My SSN is 423-91-0482. Help me file taxes."
pii = scan_prompt_for_pii(prompt)

policy_eval = arelis.governance.evaluatePolicy({
    "runId": run_id,
    "checkpoint": {
        "content": {
            "pii_detected": pii["hasPii"],
            "pii_types": [f["type"] for f in pii["findings"]],
            "pii_count": len(pii["findings"]),
        }
    },
    "policyIds": [policy_id],
})

denied = any(d.get("decision") == "deny" for d in policy_eval.get("decisions", []))

if denied:
    arelis.events.create({
        "runId": run_id,
        "eventType": "model_invocation_blocked",
        "actor": {"type": "human", "id": "demo-user"},
        "resource": {"type": "model", "id": "gemini-2.0-flash"},
        "action": "blocked_by_policy",
        "timestamp": now_iso(),
        "metadata": {"policy_id": policy_id, "pii_count": len(pii["findings"])}
    })

# 3) Risk evaluation
risk = arelis.risk.evaluate({
    "runId": run_id,
    "policyDecisions": policy_eval.get("decisions", []),
    "quotaState": {
        "audit_event": {"used": 4500, "limit": 100000},
        "compliance_proof": {"used": 23, "limit": 500},
    },
    "evaluationSignals": [
        {"name": "pii_detected", "value": 1 if pii["hasPii"] else 0, "severity": "high" if pii["hasPii"] else "low"},
    ],
})

# 4) Causal graph + commit + lineage
events = arelis.events.list({"runId": run_id, "limit": 100})
items = events.get("data", [])
if items:
    nodes = [
        {
            "id": e["eventId"],
            "type": e["eventType"],
            "data": {"action": e.get("action"), "timestamp": e.get("timestamp")},
        }
        for e in items
    ]
    edges = [
        {"source": items[i - 1]["eventId"], "target": items[i]["eventId"], "type": "sequence"}
        for i in range(1, len(items))
    ]

    arelis.replay.startCausalGraph({"runId": run_id, "nodes": nodes, "edges": edges})
    commit = arelis.graphs.commit(run_id)
    lineage = arelis.graphs.lineage(run_id, nodes[0]["id"])

    # 5) Proof generation + verification
    proof = arelis.proofs.create({
        "runId": run_id,
        "schemaVersion": "v2",
        "composed": {"layers": ["event_integrity", "causal_consistency", "policy_compliance"]},
    })
    verification = arelis.proofs.verify({"proofId": proof["proofId"]})

    print({
        "runId": run_id,
        "blocked": denied,
        "risk": risk,
        "rootHash": commit.get("rootHash"),
        "lineageNodes": len(lineage.get("nodes", [])),
        "proofVerified": verification.get("verified"),
    })

Unified Orchestration and ai_system_id

Unified APIs use snake_case ai_system_id; platform HTTP payloads/queries remain camelCase aiSystemId.

import asyncio
from arelis import (
    AgentModelResponse,
    GovernedAgentRunInput,
    GovernedAgentTool,
    GovernedInvokeInput,
    create_arelis,
)


async def main() -> None:
    arelis = create_arelis(
        {
            "platform": {
                "apiKey": "ak_live_or_test",
                "aiSystemId": "sys_platform_default",
            },
            "ai_system_id": "sys_sdk_default",
        }
    )

    await arelis.governed_invoke(
        GovernedInvokeInput(
            run_id="run_unified_1",
            model="gemini-2.5-flash",
            prompt="Summarize this ticket",
            ai_system_id="sys_per_call_override",
            invoke=lambda sanitized_prompt: f"ok:{sanitized_prompt}",
        )
    )

    await arelis.agents.run(
        GovernedAgentRunInput(
            run_id="run_agent_1",
            model="gemini-2.5-flash",
            prompt="Find order A-100",
            tools=[GovernedAgentTool(name="lookup_order")],
            invoke_model=lambda _input: AgentModelResponse(text="Order found", finish_reason="stop"),
            execute_tool_call=lambda _input: {"ok": True},
        )
    )


asyncio.run(main())

Effective AI system resolution order:

1. Per-call ai_system_id on governed_invoke(...) / agents.run(...)
2. create_arelis({"ai_system_id": ...})
3. Platform config default create_arelis({"platform": {"aiSystemId": ...}})
4. Omitted

Unified side-effect enrichment remains best-effort. Platform event writes/uploads, event listing enrichment, proof creation, risk evaluation, and graph fetch failures are returned in the result warnings field.

agents.run(...) status values are completed, blocked, max_steps_reached, and failed.

Governance Gate Helpers (Managed PII + Timings)

import asyncio
from arelis import (
    ActorRef,
    ArelisPlatform,
    EvaluatePreInvocationGateInput,
    GovernanceGateTelemetryOptions,
    ScanPromptForPiiOptions,
    WithGovernanceGateOptions,
    evaluate_pre_invocation_gate,
    scan_prompt_for_pii,
    with_governance_gate,
)


async def main() -> None:
    platform = ArelisPlatform(
        {
            "apiKey": "ak_live_or_test",
            "aiSystemId": "sys_platform_default",
        }
    )

    prompt = "Customer email is alice@example.com. Summarize request."

    managed = platform.governance.getPiiConfig({"namespace": "pii.default"})
    pii = scan_prompt_for_pii(prompt, ScanPromptForPiiOptions(redactor_config=managed))
    print(f"PII findings: {len(pii.findings)}")

    gate_input = EvaluatePreInvocationGateInput(
        run_id="run_gate_1",
        ai_system_id="sys_gateway",
        prompt=prompt,
        model="gemini-2.5-flash",
        actor=ActorRef(type="service", id="svc-gateway"),
    )

    decision = await evaluate_pre_invocation_gate(platform, gate_input)
    print(f"Decision: {decision.decision}")
    print(f"Total gate ms: {decision.metadata.timings.total_ms if decision.metadata.timings else 'n/a'}")

    guarded = await with_governance_gate(
        platform,
        gate_input,
        lambda: {"ok": True},
        options=WithGovernanceGateOptions(
            deny_mode="return",
            telemetry=GovernanceGateTelemetryOptions(enabled=True, platform=platform),
        ),
    )
    print(f"Invoked: {guarded.invoked}")


asyncio.run(main())

with_governance_gate(...) deny modes:

• throw (default): raises GovernanceGateDeniedError
• return: returns a structured deny result without invoking

When telemetry is enabled, gate events are emitted as governance.gate.evaluated and governance.gate.outcome.

Real LLM Invocation (Gemini / Claude)

Use your preferred model SDK for inference, and keep governance decisions and audit events in Arelis:

• Run PII scan + arelis.governance.evaluatePolicy(...)
• If denied: write model_invocation_blocked event and stop
• If allowed: call Gemini/Claude, then emit model.invoked and output.delivered events
• Evaluate risk with arelis.risk.evaluate(...)
• Build graph and proof using arelis.replay.startCausalGraph(...), arelis.graphs.commit(...), arelis.proofs.create(...), and arelis.proofs.verify(...)

Platform API Surface You’ll Use Most

• create_arelis_platform
• platform.governance.policies.*
• platform.governance.evaluatePolicy(...)
• platform.events.create(...), platform.events.list(...)
• platform.risk.evaluate(...)
• platform.replay.startCausalGraph(...)
• platform.graphs.commit(...), platform.graphs.lineage(...)
• platform.proofs.create(...), platform.proofs.verify(...)

Links

• PyPI: https://pypi.org/project/ai-governance-sdk/
• npm (TypeScript package): https://www.npmjs.com/package/@arelis-ai/ai-governance-sdk
• Docs: https://api.arelis.digital/docs

License

MIT