aisquare-explainability 0.1.1

Explainability SDK for tracing, graphing, and policy auditing of AI agents

AISquare Explainability SDK

An explainability and governance stack for AI agents. Captures execution traces from any Python agent (Agno, LangChain, plain Python), delivers them to a FastAPI gateway, and projects them into a Neo4j graph with RML semantic analysis and policy detection.

How it works

Your Agent Code
    │  OpenInference / OTel spans
    ▼
SDK (aisquare_explainability_sdk/)
    │  serialised to local SQLite inbox — no gateway dependency in the hot path
    ▼
InboxSweeper (background thread)
    │  POST /v1/traces/ingest
    ▼
Gateway (FastAPI — gateway/)
    ├── Structural Worker  →  Neo4j  (Run / Span / Artifact / Policy graph)
    ├── RML Extractor      →  Postgres (claims, assumptions, evidence, inference chain)
    └── Policy Detector    →  Neo4j + Postgres (PolicyCandidate nodes, crystallization loop)
         ▼
Explainability Studio UI (modular-ai-space frontend)

Repo layout

aisquare_explainability_sdk/   SDK — capture, inbox, sweeper, adapters, manual tracers
gateway/                        FastAPI ingest gateway + worker pipeline
graph/                          Neo4j Cypher and React Flow transformation helpers
examples/                       Runnable agent examples
docs/                           Guides, concepts, API reference
tests/                          Unit and integration tests

What the SDK captures

The SDK collects two layers of signal:

Auto-instrumentation (zero-code): The AgnoAdapter installs openinference-instrumentation-agno, which automatically wraps every Agno agent run, LLM call, and tool invocation as an OTel span.

Manual tracers (governance-grade): Seven context-manager tracers you can add to any Python code — framework-agnostic:

Tracer	Purpose
AgentRunTracer	Wraps a full agent run as the root span
LLMCallTracer	Records an LLM inference call with I/O and token counts
ToolCallTracer	Records a tool invocation with parameters, result, and errors
RetrievalTracer	Records a RAG retrieval with documents and scores
HumanInterventionTracer	Records a human-in-the-loop review or correction
RoutingTracer	Records a routing/delegation decision with selected and rejected paths
MemoryTracer	Records memory read/write operations

The manual tracers also provide decorators: @trace_tool and @trace_retrieval.

What the gateway adds

The gateway worker pipeline runs after each trace is ingested:

• Structural projection: Every span becomes a Run, Span, Artifact, or policy node in Neo4j with typed edges (CONTAINS, EXECUTED, PRODUCED, CONSUMED, FLAGGED_AS).
• RML extraction: GPT-4o-mini analyzes AGENT and LLM spans to produce claims, assumptions, evidence attribution, inference chain, and policy triggers. Confidence is a blended score: 40% LLM-reported + 60% deterministic structural signals.
• Policy detection: Recurring system-prompt instructions are counted across traces. When an instruction exceeds the occurrence threshold it is promoted to a PolicyCandidate node for human review and activation.

Data model

Neo4j nodes: Studio, Run, Span, Event, Artifact, Agent, Tool, Human, PolicyCandidate, Policy

Neo4j relationships: HOSTS, CONTAINS, NEXT, EXECUTED, INTERVENED, CONSUMED, PRODUCED, REFERENCES, FLAGGED_AS, PROPOSES

Postgres tables: ingest_batches, trace_states, processed_traces, rml_analyses, policy_observations, artifact_content, extraction_failures, studio_config

SDK local durability: SQLite inbox (explainability_inbox.db)

Local setup

1. Install dependencies

pip install -e ".[dev,agno]"

2. Start infrastructure

docker compose up -d

Starts Neo4j (bolt://localhost:7687), Postgres (localhost:5433), and Redis (localhost:6379).

3. Configure environment

Copy .env.example to .env and fill in the required values:

# Database
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password
POSTGRES_DSN=postgresql://postgres:postgres@localhost:5433/explainability
REDIS_URL=redis://localhost:6379/0

# Auth
ENV=development
ALLOW_TEST_AUTH_BYPASS=true

# SDK → gateway
EXPLAINABILITY_GATEWAY_URL=http://127.0.0.1:8000
EXPLAINABILITY_API_KEY=test-key

# LLM extraction
OPENAI_API_KEY=sk-...
EXTRACTOR_MODEL=gpt-4o-mini

4. Run the gateway

uvicorn gateway.main:app --host 0.0.0.0 --port 8000

Health checks:

curl http://127.0.0.1:8000/live    # → {"status":"ok"}
curl http://127.0.0.1:8000/ready   # → {"neo4j":"ok","postgres":"ok","redis":"ok"}

5. Run an example

python examples/basics/openai_chat.py

6. Start the frontend

The UI lives in the modular-ai-space repository:

cd ../modular-ai-space
npm install
npm run dev:localdev    # http://localhost:3001

SDK usage

import aisquare_explainability_sdk as sdk

# Reads EXPLAINABILITY_GATEWAY_URL and EXPLAINABILITY_API_KEY from environment
sdk.init_from_env(service_name="my-agent")

# Run your agent — Agno is auto-instrumented, all spans captured automatically
agent.print_response("...")

# IMPORTANT for short-lived scripts: flush ensures traces reach the gateway
# before the process exits. Long-running services don't need this.
sdk.flush()

Manual instrumentation (any framework)

import aisquare_explainability_sdk as sdk

sdk.init_from_env()

with sdk.AgentRunTracer(agent_name="MyAgent", run_id="abc-123") as run:
    run.set_input("User query")

    with sdk.LLMCallTracer(model="gpt-4o-mini", provider="openai") as llm:
        response = call_openai(...)
        llm.set_input_messages([{"role": "user", "content": "..."}])
        llm.set_output_messages([{"role": "assistant", "content": response}])
        llm.set_token_counts(prompt=100, completion=50)

    with sdk.RoutingTracer(decision_type="tool_selection") as rt:
        rt.set_selected("web_search", reason="Query requires fresh data")
        rt.set_rejected([{"name": "cached_search", "reason": "Cache is stale"}])

    run.set_output("Agent final answer")

sdk.flush()

Examples

File	What it shows
examples/basics/openai_chat.py	Simplest traced Agno agent
examples/agents/deep_nesting.py	3-level agent delegation team
examples/agents/multi_agent_branching.py	Multi-agent research team
examples/agents/routing_decision.py	Tool routing and conditional paths
examples/agents/tool_heavy_workflow.py	Many tool calls in one run
examples/governance/error_handling.py	Error nodes (red) in graph
examples/governance/policy_detection.py	3 runs → policy candidate detection
examples/governance/policy_workflow.py	Full detect → review → activate cycle
examples/governance/e2e_governed_workflow.py	Complete governed workflow with activation
examples/streaming/monitor_execution.py	SSE-based live execution monitoring

API surface

Ingest:

POST /v1/traces/ingest
Authorization: Bearer <api-key>
Content-Type: application/json

Studio UI reads:

GET  /v1/studios/{id}/ui/runs
GET  /v1/studios/{id}/ui/runs/{run_id}
GET  /v1/studios/{id}/ui/runs/{run_id}/graph?mode=span|event
GET  /v1/studios/{id}/ui/runs/{run_id}/xtrace
GET  /v1/studios/{id}/ui/runs/{run_id}/rml
GET  /v1/studios/{id}/ui/runs/{run_id}/nodes/{node_id}
GET  /v1/studios/{id}/ui/policies
GET  /v1/studios/{id}/ui/agents
GET  /v1/studios/{id}/ui/setup
GET  /v1/studios/{id}/artifacts/{artifact_id}

Policy management:

PATCH /v1/studios/{id}/policies/{fingerprint}/activate
PATCH /v1/studios/{id}/policy-config

Admin:

GET  /metrics
GET  /v1/admin/worker-health
GET  /v1/admin/circuit-breakers
POST /v1/admin/recover
GET  /v1/admin/extraction/failures
GET  /v1/admin/policy-candidates

Documentation

Guides:

• Getting started — full local setup walkthrough
• Quickstart (5 min) — first trace fast
• SDK usage guide — manual instrumentation patterns

Reference:

• SDK API reference — all SDK classes and functions
• REST API reference — all gateway endpoints

Concepts:

• RML — Reasoning Markup Language
• Policy system — detection and enforcement
• Graph modes — span vs event views
• Architecture — system design

Tests

python -m pytest -q
python -m ruff check .

Docker-backed integration test:

RUN_DOCKER_INTEGRATION=1 python -m pytest tests/test_integration_stack.py -q

License

MIT