Staging broke twice this month. Both times a JSON payload changed shape. Nobody noticed until a client failed. No code review caught it. This article builds a small drift detector that runs on free AI tokens and a free server. Disclosure: This article was prepared as part of MonkeyCode's product outreach. The open-source MonkeyCode project currently offers free model access and a free server option. That is enough for this experiment. Verify the current terms before you rely on them.
Why Drift Is a Trust Problem
Hand-written validators fail in a predictable way. You list the fields you expect, and then the schema changes. The rule breaks because you did not predict the new shape.
An AI classifier looks attractive. It can compare old and new payloads without predefined rules. That is also the trap. The model trusts whatever you hand it. Send the whole JSON body, and the context budget disappears into irrelevant values.
The solution is a pre-processor. Compact the payload into a fingerprint. Then let the model compare fingerprints only. This article walks through the full pipeline.
The Pipeline
The workflow has three pieces:
- Compact a JSON payload into a stable fingerprint.
- Ask a model to compare that fingerprint with a stored baseline.
- Store the decision in SQLite for later review.
The fingerprint is the original artifact here. It is smaller, deterministic, and removes most customer data before the model sees it.
Step 1: Build a Compact Fingerprint
This function extracts keys, types, and short value hints. It discards long strings and timestamps. It is a pre-processor, not a model.
import json
from collections import OrderedDict
def fingerprint(obj, max_sample=40):
if isinstance(obj, dict):
return {
"type": "object",
"fields": OrderedDict(
(k, fingerprint(v, max_sample)) for k, v in obj.items()
),
}
if isinstance(obj, list):
return {"type": "array", "sample": [fingerprint(v, max_sample) for v in obj[:2]], "count": len(obj)}
if isinstance(obj, str):
return {"type": "string", "hint": obj[:max_sample]}
if isinstance(obj, bool):
return {"type": "boolean", "value": obj}
if isinstance(obj, (int, float)):
return {"type": "number", "value": obj}
return {"type": "null"}
Example output for a small payload:
{"type":"object","fields":{"id":{"type":"number","value":4},"name":{"type":"string","hint":"Ada"},"items":{"type":"array","sample":[],"count":0}}}
That compact blob fits in a small context window. It also makes the comparison less noisy, because the model focuses on shape, not exact values.
Step 2: Ask the Model to Compare
The prompt below is unexecuted pseudocode. Use it as a starting point. The goal is constrained JSON output, not free-form prose.
SYSTEM = '''
You compare a JSON baseline to a new JSON fingerprint.
Return JSON with:
label: NO_CHANGE | MINOR_CHANGE | BREAKING_CHANGE
reason: one sentence
changed_fields: list
NO_CHANGE: same fields and compatible types.
MINOR_CHANGE: new optional-looking field or type widening.
BREAKING_CHANGE: required field removed, type changed, or name changed.
'''
Pass the baseline and the new fingerprint as two small blobs. The model sees a comparison problem, not a data dump. Constrained output is important here. A JSON response is easy to store, query, and test.
Step 3: Store Every Decision
A SQLite table is enough. Append instead of overwriting, because history is the audit trail.
CREATE TABLE IF NOT EXISTS decisions (
id INTEGER PRIMARY KEY,
checked_at TEXT DEFAULT CURRENT_TIMESTAMP,
label TEXT NOT NULL,
reason TEXT,
changed_fields TEXT
);
Every scheduled run inserts one row. A morning-after inspection answers one question: what drifted last night?
What Each Label Means
| Label | Meaning | Next step |
|---|---|---|
| NO_CHANGE | Nothing suspicious | Archive quietly |
| MINOR_CHANGE | New field or wider type | Review queue for a human |
| BREAKING_CHANGE | Field removed or type changed | Alert channel immediately |
This table is the part you can test with a week of historical payloads. The model is not the source of truth. Human review still decides.
A Cron Job Is the Whole Server
The workload is tiny. One fingerprint per queue item. One comparison every few minutes. One row in SQLite. A free server option is genuinely enough.
Sample cron entry on a small Linux box:
*/10 * * * * cd /opt/drift && python drift_job.py >> /var/log/drift.log 2>&1
If the job fails, the log shows a stack trace next to the last decision. That is enough for a first deployment. You do not need a queue, a database cluster, or a managed platform.
Evaluate Before You Trust the Label
Build an evaluation set of fifty real payload pairs. Mark each pair by hand with one of the three labels. Then compare the pipeline output to a keyword baseline that checks field presence.
The simplest baseline flags a change when a key appears in one payload but not in the other. The AI should do better at type-change detection. If it does not, tighten the prompt or the fingerprint.
def evaluate(gold, predicted):
return sum(1 for g, p in zip(gold, predicted) if g == p) / len(gold)
Run this evaluation before scheduling the cron job. An untested drift detector gives you false safety.
Limitations and Who Should Not Use This
This approach is not for strict compliance workloads. If the payload contains regulated data, keep it on your own hardware. Free tiers can change quotas, models, or availability without notice. The pipeline must survive losing the provider.
Skip this if you process only a few dozen payloads per week. A manual diff is cheaper. Skip it too if your schema changes are already enforced by a typed gateway.
The value is in the process, not in the free allowance.
Try the Experiment
Start with a hundred historical payloads from one queue. Build fingerprints, label fifty pairs manually, then check model agreement. If the labels are stable, schedule the job.
The free model access and free server options in the open-source MonkeyCode project can support this experiment today. Use them to test the workflow. Re-run the evaluation whenever the terms or the model line change.
Top comments (0)