Your agent ran fine for three weeks. Then it started making subtle errors — wrong API calls, hallucinated responses, skipping validation steps. By the time you noticed, it had corrupted a week of data.
This is the agent drift problem. It's not dramatic. It's not a crash. It's a slow degradation that looks like success until it isn't.
I built four tools that detect drift before it costs you. Here's what each does and how to use them.
1. Agent Drift Detector — Strategy Space Monitoring
Agents optimize for the wrong thing when their strategy space narrows. This tool tracks the diversity of tool calls, decision patterns, and output distributions over time.
# drift_detector.py
import json
from collections import Counter
from datetime import datetime, timedelta
class DriftDetector:
def __init__(self, window_hours=24, diversity_threshold=0.6):
self.window = timedelta(hours=window_hours)
self.threshold = diversity_threshold
self.history = []
def record_action(self, agent_id: str, action_type: str, params: dict):
signature = f"{action_type}:{hash(frozenset(params.items()))}"
self.history.append({
"agent_id": agent_id,
"signature": signature,
"timestamp": datetime.now()
})
self._prune()
def _prune(self):
cutoff = datetime.now() - self.window
self.history = [h for h in self.history if h["timestamp"] > cutoff]
def get_diversity_score(self, agent_id: str) -> float:
agent_actions = [h["signature"] for h in self.history if h["agent_id"] == agent_id]
if not agent_actions:
return 1.0
counts = Counter(agent_actions)
# Normalized entropy
total = len(agent_actions)
entropy = -sum((c/total) * (c/total).bit_length() for c in counts.values())
max_entropy = (len(counts)).bit_length() if len(counts) > 1 else 1
return entropy / max_entropy
def is_drifting(self, agent_id: str) -> bool:
return self.get_diversity_score(agent_id) < self.threshold
# Usage
detector = DriftDetector()
detector.record_action("agent-1", "api_call", {"endpoint": "/users", "method": "GET"})
print(f"Drifting: {detector.is_drifting('agent-1')}")
Alert trigger: When diversity drops below 0.6 for 2+ hours, the agent is exploiting a narrow pattern — likely overfitting.
2. Consequence Accountability Ledger — Decision Receipts
Every agent decision gets a cryptographic receipt. When something goes wrong, you have an immutable audit trail — not logs that can be rotated or deleted.
# consequence_ledger.py
import hashlib
import json
import time
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class DecisionReceipt:
agent_id: str
decision: str
context_hash: str
expected_outcome: str
consequence_score: float # 0-100, higher = more impact
timestamp: float
prev_receipt_hash: Optional[str] = None
def to_hash(self) -> str:
data = {k: v for k, v in asdict(self).items() if k != 'prev_receipt_hash'}
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
class ConsequenceLedger:
def __init__(self):
self.chain = []
def record(self, receipt: DecisionReceipt) -> str:
receipt.prev_receipt_hash = self.chain[-1].to_hash() if self.chain else "0" * 64
receipt_hash = receipt.to_hash()
self.chain.append(receipt)
return receipt_hash
def verify_chain(self) -> bool:
for i, r in enumerate(self.chain):
expected_prev = self.chain[i-1].to_hash() if i > 0 else "0" * 64
if r.prev_receipt_hash != expected_prev:
return False
return True
# Usage
ledger = ConsequenceLedger()
ledger.record(DecisionReceipt(
agent_id="trading-bot",
decision="execute_buy",
context_hash=hashlib.sha256(b"BTC>65k,RSI<30").hexdigest(),
expected_outcome="profit > 2%",
consequence_score=85,
timestamp=time.time()
))
print(f"Chain valid: {ledger.verify_chain()}")
Key insight: Agents with high consequence scores but low verification rates are your biggest risk.
3. Cost Ceiling Enforcer — Budget Guardrails
Agents in retry loops can 50x your inference costs in minutes. This enforces hard ceilings per agent, per task, per day.
# cost_ceiling.py
import time
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict
@dataclass
class Budget:
daily_limit: float
task_limit: float
per_step_limit: float
spent_today: float = 0
spent_task: float = 0
spent_step: float = 0
last_reset: float = time.time()
class CostCeiling:
def __init__(self):
self.budgets: Dict[str, Budget] = {}
def get_budget(self, agent_id: str, defaults=None) -> Budget:
if agent_id not in self.budgets:
self.budgets[agent_id] = Budget(**(defaults or {
"daily_limit": 50.0,
"task_limit": 10.0,
"per_step_limit": 2.0
}))
b = self.budgets[agent_id]
if time.time() - b.last_reset > 86400:
b.spent_today = 0
b.last_reset = time.time()
return b
def can_spend(self, agent_id: str, amount: float, task_id: str = None) -> bool:
b = self.get_budget(agent_id)
if b.spent_today + amount > b.daily_limit:
return False
if b.spent_step + amount > b.per_step_limit:
return False
if task_id and b.spent_task + amount > b.task_limit:
return False
return True
def charge(self, agent_id: str, amount: float, task_id: str = None):
b = self.get_budget(agent_id)
b.spent_today += amount
b.spent_step += amount
if task_id:
b.spent_task += amount
def reset_step(self, agent_id: str):
self.get_budget(agent_id).spent_step = 0
# Usage
ceiling = CostCeiling()
for step in range(10):
if not ceiling.can_spend("agent-1", 0.05):
print("CEILING HIT - stopping")
break
ceiling.charge("agent-1", 0.05)
ceiling.reset_step("agent-1")
4. Motivation Decay Detector — Plateau Alerting
Agents stop exploring. They find a local optimum and exploit it forever. This tracks exploration vs exploitation ratio.
# motivation_decay.py
from collections import deque
import random
class MotivationTracker:
def __init__(self, window=100, min_exploration=0.15):
self.window = window
self.min_exploration = min_exploration
self.actions = deque(maxlen=window)
def record(self, action_type: str, is_novel: bool):
self.actions.append({"type": action_type, "novel": is_novel, "ts": time.time()})
def exploration_rate(self) -> float:
if not self.actions:
return 1.0
novel = sum(1 for a in self.actions if a["novel"])
return novel / len(self.actions)
def is_stagnant(self) -> bool:
return self.exploration_rate() < self.min_exploration
def plateau_duration(self) -> int:
"""Seconds since exploration rate dropped below threshold"""
for a in reversed(self.actions):
if a["novel"]:
return int(time.time() - a["ts"])
return int(time.time() - self.actions[0]["ts"]) if self.actions else 0
# Usage
tracker = MotivationTracker()
for i in range(50):
tracker.record("api_call", is_novel=(i < 10)) # First 10 novel, then exploit
print(f"Exploration: {tracker.exploration_rate():.2%}")
print(f"Stagnant: {tracker.is_stagnant()}")
print(f"Plateau: {tracker.plateau_duration()}s")
The Pattern: Detect → Alert → Intervene
All four tools share a philosophy: measure the invisible, alert before failure, make intervention trivial.
| Tool | Measures | Alert Threshold | Intervention |
|---|---|---|---|
| Drift Detector | Strategy diversity | < 0.6 for 2hrs | Force exploration |
| Consequence Ledger | Decision impact | High score, low verify | Require sign-off |
| Cost Ceiling | Spend velocity | 80% of limit | Hard stop |
| Motivation Decay | Novelty rate | < 15% for 1hr | Inject randomness |
Want These Running in Production?
Full catalog of my AI agent tools at https://thebookmaster.zo.space/bolt/market
Each tool is available as a standalone API with:
- Real-time dashboards
- Webhook alerts (Slack, email, Telegram)
- Historical trend analysis
- Multi-agent fleet support
Built these because I got tired of finding out my agents drifted three weeks too late. What's the silent failure mode you're most worried about?
Top comments (0)