DriftGuard

DriftGuard is a semantic mistake-memory and guardrail layer for autonomous agents.

It sits between intent and execution, allowing agents to learn from past failures and avoid repeating them.

DriftGuard stores structured causal memories:

action → feedback → outcome

and surfaces warnings when similar risky actions appear again.

It also keeps a separate success memory: when an action resembles a past action that led to a good outcome, DriftGuard surfaces a positive reinforcement/recommendation alongside (or instead of) warnings.

It works with:

• MCP agents
• LangGraph workflows
• custom Python agents
• tool-calling planners
• autonomous pipelines

---

Why DriftGuard Exists

Agents today can act.

They usually cannot remember mistakes meaningfully.

Typical failure loop:

agent makes mistake
agent retries
agent repeats mistake
agent retries again

DriftGuard introduces a semantic failure memory layer:

plan step
↓
DriftGuard review
↓
warning surfaced
↓
agent revises action

This improves:

• stability
• reliability
• convergence speed
• evaluation consistency
• production safety

without requiring changes to your planner architecture.

---

What DriftGuard Does

DriftGuard provides:

• semantic mistake memory • semantic success memory with positive reinforcement • similarity-aware warning and reinforcement retrieval • policy-based execution guardrails • merge + deduplicate memory graphs • JSON, SQLite, or Postgres persistence • runtime metrics • pruning of stale weak memories • MCP server integration • LangGraph adapters • offline benchmark harness

---

Installation

Install from PyPI:

pip install driftguard-ai

Install test dependencies:

pip install "driftguard-ai[test]"

Install LangGraph demo dependencies:

pip install "driftguard-ai[demo]"

Install the spaCy normalization model:

python -m spacy download en_core_web_sm

---

Quick Example (Python Agent)

from driftguard import DriftGuard

guard = DriftGuard()

review = guard.before_step("increase salt")

if review.warnings:
    print(review.warnings[0].risk)

if review.reinforcements:
    print(review.reinforcements[0].recommendation)

guard.record(
    action="increase salt",
    feedback="too salty",
    outcome="dish ruined",
)

DriftGuard now remembers this failure and warns on similar steps later.

---

Positive Reinforcement (Success Memory)

Alongside the mistake graph, DriftGuard maintains a second, independent success graph. Recording a successful action → feedback → outcome chain lets DriftGuard recommend that action again when a similar one is proposed later:

guard.record_success(
    action="add more salt",
    feedback="well seasoned",
    outcome="dish praised",
)

review = guard.before_step("add a pinch of salt")

for reinforcement in review.reinforcements:
    print(reinforcement.trigger, "->", reinforcement.recommendation)

Each Reinforcement has the same shape as a Warning, mirrored for positive outcomes:

trigger: str          # the past successful action
recommendation: str   # the positive feedback it produced
frequency: int
confidence: float

The mistake and success graphs are completely separate stores — they use their own merge/prune passes (via shared engines) and their own persistence files/tables, so recording a mistake never affects the success graph and vice versa.

---

Guard Policies

Control how the agent reacts to detected risks:

from driftguard import DriftGuard, DriftGuardSettings

guard = DriftGuard(
    settings=DriftGuardSettings(
        guard_policy="acknowledge",
        guard_min_confidence=0.8,
    )
)

Supported modes:

policy	behavior
warn	surface warning only
block	raise exception
acknowledge	require confirmation
record_only	store memory but skip review

---

MCP Server Usage

Run DriftGuard as an MCP server:

driftguard-mcp

Available tools:

register_mistake
register_success
query_memory
deep_prune
graph_stats
guard_metrics

query_memory returns both warnings (from the mistake graph) and reinforcements (from the success graph) for the given context.

Example Claude Desktop config:

{
  "mcpServers": {
    "driftguard": {
      "command": "driftguard-mcp"
    }
  }
}

---

LangGraph Integration

Create a review node inside a LangGraph workflow:

from driftguard import DriftGuard
from driftguard import make_langgraph_review_node

guard = DriftGuard()

review_node = make_langgraph_review_node(guard)

Drop this node directly into a planner graph.

---

Generic Payload Adapter

Review arbitrary planner payloads:

from driftguard import DriftGuard, review_payload

guard = DriftGuard()

result = review_payload(
    guard,
    {"action": "increase salt", "attempt": 2},
)

---

CLI Benchmark Tool

Evaluate merge and retrieval quality:

driftguard-benchmark

Export structured results:

driftguard-benchmark --format json

Measures:

• merge precision
• merge recall
• retrieval precision
• retrieval recall
• F1 score

---

Storage Model

DriftGuard uses:

in-memory semantic graph runtime
+
persistent storage backend

Supported persistence:

backend	purpose
JSON	local experiments
SQLite	production workflows
Postgres	shared / multi-process deployments

Example configuration:

from driftguard import DriftGuardSettings

settings = DriftGuardSettings(
    storage_backend="sqlite",
    sqlite_filepath="driftguard.sqlite3",
)

The Postgres backend requires the optional postgres extra:

pip install "driftguard-ai[postgres]"

from driftguard import DriftGuardSettings

settings = DriftGuardSettings(
    storage_backend="postgres",
    postgres_dsn="postgresql+psycopg://user:pass@host:5432/driftguard",
)

Embeddings are stored as JSONB on Postgres and plain JSON elsewhere; the schema (driftguard_meta, driftguard_nodes, driftguard_edges) is created automatically on first save.

The mistake graph and the success graph are persisted independently, using the same backend:

backend	mistake graph	success graph
JSON	driftguard_graph.json	driftguard_success_graph.json
SQLite	driftguard_graph.sqlite3	driftguard_success_graph.sqlite3
Postgres	driftguard_meta/nodes/edges	success_driftguard_meta/nodes/edges

For Postgres, both table sets live in the same database/DSN — the success graph simply uses a success_ table prefix.

---

Metrics and Observability

Runtime metrics available:

from driftguard import build_runtime

runtime = build_runtime()

snapshot = runtime.metrics_snapshot()

print(snapshot["counters"])

Includes:

reviews
warnings
blocks
acknowledgements
records
node reuse
edge reuse
prune activity

Also available via MCP:

guard_metrics

---

Example Architecture Placement

Typical agent loop:

planner
 ↓
candidate action
 ↓
DriftGuard review
 ↓
warning surfaced
 ↓
planner revision
 ↓
execution
 ↓
feedback recorded

DriftGuard improves stability without replacing the planner.

---

Local Demos

Two included demos:

Rule-based simulator

Offline deterministic walkthrough:

python demo/rule_based/demo_agent.py

Shows:

• merge behavior
• warning retrieval
• pruning cleanup
• graph evolution

---

LangGraph LLM agent demo

pip install "driftguard-ai[demo]"
python demo/langgraph/demo_agent.py

Demonstrates:

planner → guard → revise → execute loop

with real model interaction.

---

CLI Entry Points

Installed automatically:

driftguard-mcp
driftguard-benchmark

---

Configuration Surface

Example advanced setup:

from driftguard import DriftGuardSettings

settings = DriftGuardSettings(
    retrieval_top_k=5,
    retrieval_min_similarity=0.60,
    similarity_threshold_action=0.72,
    guard_policy="warn",
)

The success graph has its own filepaths, separate from the mistake graph:

settings = DriftGuardSettings(
    graph_filepath="driftguard_graph.json",
    success_graph_filepath="driftguard_success_graph.json",
    sqlite_filepath="driftguard.sqlite3",
    success_sqlite_filepath="driftguard_success.sqlite3",
)

Full configuration supports:

retrieval tuning
similarity thresholds
guard policy modes
storage backend selection
embedding configuration
graph pruning controls
logging verbosity

---

When To Use DriftGuard

DriftGuard helps when your agent:

• retries failing steps repeatedly
• forgets past execution errors
• needs execution-time guardrails
• requires semantic mistake recall
• runs multi-step planners
• uses LangGraph or MCP
• executes tools autonomously

---

Project Status

Current release includes:

• semantic merge engine
• similarity retrieval engine
• dual mistake/success memory graphs with positive reinforcement
• graph persistence layer
• SQLite and Postgres backends
• MCP server
• LangGraph adapter
• benchmark harness
• runtime metrics
• pruning engine
• deterministic demo runtime
• pytest coverage

DriftGuard is suitable for early production experimentation and agent-infrastructure research workflows.

---

License

MIT License