Skip to main content

AI Governance SDK for Python — governed AI orchestration

Project description

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", "http://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.

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

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

ai_governance_sdk-2.0.4.tar.gz (326.3 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

ai_governance_sdk-2.0.4-py3-none-any.whl (280.9 kB view details)

Uploaded Python 3

File details

Details for the file ai_governance_sdk-2.0.4.tar.gz.

File metadata

  • Download URL: ai_governance_sdk-2.0.4.tar.gz
  • Upload date:
  • Size: 326.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for ai_governance_sdk-2.0.4.tar.gz
Algorithm Hash digest
SHA256 cc7452e9a280d6b4d3f8fb2bc0f0b418c3d07153ec750d6796501e9e5c55768b
MD5 b8bbfffb02bde3a4e7fe17e4fc81e57c
BLAKE2b-256 862f87ae089d9e6e7beb9b1e1c68aa97f8ca76c6d13bd78561aee90a277c9392

See more details on using hashes here.

File details

Details for the file ai_governance_sdk-2.0.4-py3-none-any.whl.

File metadata

File hashes

Hashes for ai_governance_sdk-2.0.4-py3-none-any.whl
Algorithm Hash digest
SHA256 5c39404aa23237ed66a5662e201daef2f35573e95c419dea270989d1541f9ec4
MD5 1b66cf90525fab6af4a5e8b05b52208d
BLAKE2b-256 d9affb12a5d288a1929276db5b4da648faa2e52d81f05db9b482c5b85aadbdc3

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page