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