Have you ever built an intelligent agent, perhaps one that uses Q-Learning on dynamic text streams, only to discover it's making nonsensical decisions because an upstream API is subtly misbehaving? I certainly have. It’s not always an outright 500 error that brings down a system; sometimes, it’s the insidious 200 OK with a malformed payload, a sudden drop in result count, or an unexpected shift in content structure – what I call *content drift*. For those of us architecting adaptive systems that rely on the freshness and integrity of dynamic API data, these subtle anomalies are silent killers. This post will walk you through building a robust, adaptive anomaly detection system using Isolation Forest with a sliding window, designed to catch these elusive deviations in real-time, preventing cascading failures and ensuring your intelligent agents operate on trustworthy data.
Key Takeaways
- Adaptive anomaly detection for API streams requires monitoring both performance metrics (latency) and content characteristics (result count, structural integrity).
- Isolation Forest is effective for unsupervised anomaly detection in multivariate data streams, identifying outliers based on feature isolation.
- A sliding window approach, combined with periodic model retraining, allows anomaly detectors to adapt to evolving data distributions without requiring full historical retraining.
- Feature engineering for API responses should capture not just direct values but also structural and statistical properties of the payload.
- Operationalizing anomaly scores involves setting dynamic thresholds and providing contextual information to make alerts actionable.
The Problem: The Silent Killer of Adaptive Agents
In our previous explorations into architecting adaptive decision-making agents, we've focused on how these systems learn and react to dynamic information. What we often overlook, however, is the fragility of the data sources themselves. My team once deployed an agent that processed book metadata from a public API to inform content recommendations. Everything worked perfectly for months, then suddenly, the recommendations started to degrade. The API wasn't down; it was returning 200 OKs, but the number of books for common queries had inexplicably dropped from hundreds to single digits, and critical fields like `author_name` were intermittently missing. Our static health checks caught nothing. This *content drift* silently poisoned our agent's decision-making process, leading to a poor user experience and wasted compute. We needed a system that could detect these subtle, multivariate shifts – not just network issues, but data integrity issues – in real-time.
Data and Sources
For this demonstration, we'll simulate interactions with the Open Library Search API. This public API provides a rich source of book metadata, perfect for illustrating content drift.
Data accessed on 2024-07-29.
Step 1 — Synthesizing a Production API Stream and Feature Engineering
The first sub-problem is to simulate a continuous stream of API interactions and extract meaningful, quantifiable features from each response. Raw JSON is not directly consumable by most ML models, so we need to transform it into a structured, numerical time series. This involves measuring response time, parsing the JSON payload, and extracting key characteristics like the number of results, the average length of titles, and the presence of critical fields. To make our simulation realistic, I'll introduce both latency spikes and content drift (e.g., reduced result count or missing fields) at specific intervals.
Our `fetch_and_feature` function will handle the API call, timing, and feature extraction. We'll also simulate different "states" of the API to represent normal behavior, high latency, and content drift.
import time
import requests
import pandas as pd
import numpy as np
import json
from sklearn.ensemble import IsolationForest
from collections import deque
# Configuration
API_URL = "https://openlibrary.org/search.json?q=data+science"
WINDOW_SIZE = 100 # Number of recent samples to keep for the sliding window
TRAINING_INTERVAL = 50 # Retrain Isolation Forest every N samples
ANOMALY_THRESHOLD = -0.15 # Isolation Forest decision function threshold
def fetch_and_feature(api_url: str, simulate_anomaly_type: str = None) -> dict:
"""
Fetches data from the API and extracts features, with optional anomaly simulation.
Features include latency, number of documents, and average title length.
"""
start_time = time.time()
try:
# Simulate latency spikes
if simulate_anomaly_type == "latency":
time.sleep(np.random.uniform(0.5, 1.5)) # Introduce significant delay
elif simulate_anomaly_type == "minor_latency":
time.sleep(np.random.uniform(0.1, 0.3)) # Minor delay
response = requests.get(api_url, timeout=5) # 5-second timeout for real-world scenarios
latency = time.time() - start_time
response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
data = response.json()
num_docs = len(data.get('docs', []))
# Simulate content drift: fewer results or missing fields
if simulate_anomaly_type == "content_drift_count":
num_docs = max(0, num_docs // 5) # Drastically reduce doc count
elif simulate_anomaly_type == "content_drift_structure":
# Simulate missing 'title' or 'author_name' for some results
if 'docs' in data and len(data['docs']) > 0:
for i in range(min(3, len(data['docs']))): # Affect first 3 docs
if 'title' in data['docs'][i]:
del data['docs'][i]['title'] # Simulate missing title
avg_title_len = 0
if num_docs > 0:
titles = [doc.get('title', '') for doc in data.get('docs', [])]
# Ensure titles are strings before calculating length
titles = [str(t) for t in titles if t is not None]
if titles:
avg_title_len = sum(len(t) for t in titles) / len(titles)
# Feature for consistency: check if 'author_name' exists in first few docs
has_author_name = 1 if any('author_name' in doc for doc in data.get('docs', [])[:5]) else 0
return {
'timestamp': pd.to_datetime(time.time(), unit='s'),
'latency': latency,
'num_docs': num_docs,
'avg_title_len': avg_title_len,
'has_author_name': has_author_name,
'anomaly_sim_type': simulate_anomaly_type if simulate_anomaly_type else 'none'
}
except requests.exceptions.Timeout:
latency = time.time() - start_time
print(f"[{pd.to_datetime(time.time(), unit='s')}] API Timeout after {latency:.2f}s")
return {
'timestamp': pd.to_datetime(time.time(), unit='s'),
'latency': latency,
'num_docs': 0, # Assume 0 docs on timeout
'avg_title_len': 0,
'has_author_name': 0,
'anomaly_sim_type': 'timeout'
}
except requests.exceptions.RequestException as e:
latency = time.time() - start_time
print(f"[{pd.to_datetime(time.time(), unit='s')}] API Request Error: {
