driftmonitor-ai 0.1.0

Real-time behavioral drift detection for AI systems

Drift//Monitor

Real-time behavioral drift detection for AI systems.

---

The Problem

LLM outputs degrade over long sessions. Tone shifts. Goals wander. Instructions erode. Content diverges from source material. This happens silently -- no existing tool monitors for it in real time.

Drift//Monitor is the andon cord for language models. It sits as a transparent proxy between your application and the LLM API, measuring output deviation from an established baseline using embedding distance and statistical process control.

Quick Start

pip install driftmonitor

import anthropic
from driftmonitor import DriftAwareClient, SessionConfig

# Wrap your existing Anthropic client -- zero code changes needed
base = anthropic.Anthropic()
client = DriftAwareClient(base, config=SessionConfig(name="My Session"))

# Use exactly like the base client
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)
# Response passes through unchanged. Drift scores logged to SQLite.

# Check drift
print(client.latest_score)        # DriftScore(persona=5.2, goal=3.1, ...)
print(client.latest_severity)     # Severity.NOMINAL
print(client.composite_history)   # [2.1, 3.4, 5.2, ...]

How It Works

Three-Layer Architecture

Layer	Function	Technology
Capture	Intercept every LLM input/output via proxy SDK	Drop-in Python client wrapper
Score	Embed + measure statistical deviation from anchor	sentence-transformers, SPC control charts
Act	Alert, correct, gate, and log drift events	FastAPI, SQLite, React dashboard

Four Drift Axes

Axis	Measures	Example Signal
Persona / Tone	Voice, formality, character consistency	Technical writer starts editorializing
Goal Alignment	Adherence to original objective	Code agent starts refactoring instead of fixing
Instruction Follow	Respect for earlier constraints	Agent told not to delete files begins deleting
Semantic Fidelity	Alignment with source material	Summary contradicts source document

Severity Scale (SPC-based)

Level	Threshold	Response
NOMINAL	< mean + 1σ	Log only
LOW	≥ mean + 1σ	Dashboard flag
ELEVATED	≥ mean + 2σ	Operator notification, correction generated
CRITICAL	≥ mean + 3σ	Andon cord -- gate or halt

Operational Modes

from driftmonitor import DriftAwareClient, SessionConfig

# OBSERVE -- log everything, no intervention (default)
client = DriftAwareClient(base, config=SessionConfig(mode="observe"))

# ALERT -- notify operator on threshold breach
client = DriftAwareClient(base, config=SessionConfig(mode="alert"))

# GATE -- hold response until operator approves
client = DriftAwareClient(base, config=SessionConfig(mode="gate"))
response = client.messages.create(...)
if client.is_gated:
    client.gate_approve(operator="admin", reason="Acceptable drift")
    # or: client.gate_reject(operator="admin", reason="Too much drift")

REST API

# Install server dependencies
pip install driftmonitor[server]

# Launch the API
driftmonitor serve --port 8000

# Or with a custom database
driftmonitor serve --db /path/to/drift.db

Endpoints

Method	Path	Description
GET	/health	Service status
GET	/sessions	List all sessions
GET	/sessions/{id}	Session details
DELETE	/sessions/{id}	Delete session
GET	/sessions/{id}/events	Drift events (filterable)
GET	/sessions/{id}/latest	Most recent event
GET	/sessions/{id}/summary	Aggregate statistics
GET	/audit	Audit log entries
GET	/audit/export/json	Export audit as JSON
GET	/audit/export/csv	Export audit as CSV
WS	/ws/{session_id}	Live score streaming

CLI

# List sessions
driftmonitor sessions --db drift.db

# View scores for a session
driftmonitor scores <session-id> --limit 20

# Session summary
driftmonitor summary <session-id>

# Launch API server
driftmonitor serve --port 8000

# Version
driftmonitor version

Audit Log

Every drift event, mode change, gate decision, and correction is recorded in an immutable audit log with timestamps and operator attribution.

# Export for compliance
audit_json = client.audit.export_json(session_id="sess-001")
audit_csv = client.audit.export_csv(session_id="sess-001")

# Query entries
entries = client.audit.get_entries(action="gate_released", limit=50)

Custom Anchor

By default, the first assistant response becomes the drift baseline. You can provide a reference document instead:

client.set_anchor("Your reference text or system prompt here.")

Architecture

Your App
    |
    v
DriftAwareClient  ───>  EmbeddingEngine  ───>  SPCEngine
    |                    (sentence-transformers)  (control limits)
    |                         |                       |
    v                         v                       v
Anthropic API            DriftStore              DriftExplainer
    |                    (SQLite WAL)            (Claude Haiku)
    v                         |                       |
Response                      v                       v
(unchanged)              Audit Log              Corrections

Competitive Landscape

Category	Tools	Gap
LLM Observability	LangSmith, Helicone, Arize	Log tokens and latency. No behavioral drift scoring.
AI Testing	PromptFoo, Braintrust	Snapshot evaluation. Not live longitudinal monitoring.
AI Guardrails	Guardrails AI, NeMo	Input/output filters. Not drift detection over time.
AI Governance	Various	No operator-controlled correction with audit trail.

Development

# Clone and install
git clone https://github.com/AreteDriver/drift-monitor.git
cd drift-monitor
python -m venv .venv && source .venv/bin/activate
pip install -e ".[server,dev]"

# Run tests
pytest tests/ -v

# Lint
ruff check src/ tests/
ruff format src/ tests/

License

MIT

---

Built by AreteDriver -- standard work for AI behavior.