The model stopped returning valid JSON

You ask the model to return JSON, you parse json.loads(...) — and one day you get a JSONDecodeError. Once — that happens. But when the parse-failure rate jumps from 0.5% to 30%, it’s not random: the provider shipped a new model version, or your own system-prompt edit broke the format. Meanwhile users see a blank screen or a 500, and you learn about it from tickets.

This is not schema-drift: there the JSON is valid and parses, but the fields shifted. Here the failure is cruder — the response doesn’t parse at all (the model wrapped the JSON in ```json fences, added “Here’s your answer:”, got cut off by max_tokens) or fails schema validation.

1. Rolling window: failure rate + one alert per spike

We wrap the parse step: json.loads plus optional validation (pydantic / jsonschema). The result — success/fail — goes into a ring window of the last N. When the failure rate crosses a threshold, we push. A flag file guarantees one alert per spike, not a push on every call.

import os, json, time, requests
from collections import deque
from jsonschema import validate, ValidationError

WINDOW   = 200          # how many recent calls we keep
THRESHOLD = 0.15        # >15% failures in the window -> alert
FLAG      = "/tmp/json-parse-fail.flag"

_recent = deque(maxlen=WINDOW)   # 1 = fail, 0 = ok

SCHEMA = {                        # optional: description of the expected JSON
    "type": "object",
    "required": ["intent", "confidence"],
    "properties": {
        "intent":     {"type": "string"},
        "confidence": {"type": "number"},
    },
}

def parse_structured(raw: str):
    # Returns (obj, None) on success or (None, reason) on failure.
    try:
        obj = json.loads(raw)
    except json.JSONDecodeError as e:
        return None, f"json.loads: {e}"
    try:
        validate(obj, SCHEMA)     # drop this if you don't need validation
    except ValidationError as e:
        return None, f"schema: {e.message}"
    return obj, None

def observe(raw: str):
    obj, reason = parse_structured(raw)
    _recent.append(0 if reason is None else 1)

    if len(_recent) >= WINDOW:
        rate = sum(_recent) / len(_recent)
        if rate >= THRESHOLD and not os.path.exists(FLAG):
            open(FLAG, "w").write(str(time.time()))
            notify("🧩 JSON перестал парситься",
                   f"Доля парс-фейлов: {rate:.0%} за последние {WINDOW} вызовов.\n"
                   f"Последняя причина: {reason}\n"
                   "Проверьте: релиз модели? правка system-промпта?",
                   priority=8)
        elif rate < THRESHOLD * 0.5 and os.path.exists(FLAG):
            os.remove(FLAG)       # spike subsided — clear the flag, await the next one

    return obj, reason

def notify(title, message, priority):
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title": title, "message": message, "priority": priority},
                  timeout=5)

The threshold is on the rate, not on an absolute count — otherwise on high traffic the alert fires even at a healthy 0.5% failures. The flag clears when the rate drops to half the threshold: that way the next spike wakes you again.

2. Show WHAT the model actually returned (redacted sample)

Knowing the “JSON broke” isn’t enough — you need to see how it broke. Most often it’s ```json fences or an intro phrase before the object. We put a truncated, sanitized sample of the raw response into the push:

import re

def redact_sample(raw: str, limit: int = 400) -> str:
    # mask obvious PII and trim length so we don't drag extra data into the push
    s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "<email>", raw)
    s = re.sub(r"\b\d{12,}\b", "<num>", s)
    return s[:limit] + ("…" if len(s) > limit else "")

def observe_with_sample(raw: str):
    obj, reason = observe(raw)
    if reason and os.path.exists(FLAG):
        notify("🧩 Сырой ответ вместо JSON",
               f"Причина: {reason}\n\nЧто вернула модель:\n{redact_sample(raw)}",
               priority=7)
    return obj, reason

Often the sample gives you the diagnosis in a second: you see Here's your JSON:\n```json — meaning the prompt stopped enforcing “object only, no wrappers”. The simple fix is to strip fences before parsing:

def strip_fences(raw: str) -> str:
    m = re.search(r"```(?:json)?\s*(\{.*\}|\[.*\])\s*```", raw, re.S)
    return m.group(1) if m else raw

3. Daily baseline: today vs yesterday

The rolling window catches an acute spike; a daily compare catches slow drift. Once a day we compute the day’s failure rate and compare it to yesterday’s:

STATE = "/tmp/json-parse-daily.json"

def daily_report(ok_today: int, fail_today: int):
    total = ok_today + fail_today or 1
    rate  = fail_today / total
    prev  = (json.load(open(STATE)) if os.path.exists(STATE) else {}).get("rate", rate)

    if rate - prev >= 0.05:      # +5 pp over yesterday
        notify("🧩 Парс-фейлы растут день ко дню",
               f"Сегодня: {rate:.1%} ({fail_today}/{total})\n"
               f"Вчера:   {prev:.1%}\n"
               "Проверьте историю деплоев и релизы моделей.",
               priority=7)

    json.dump({"rate": rate, "ts": time.time()}, open(STATE, "w"))

Run the report from cron / a systemd timer or from a scheduled cloud function on YC — that’s also a handy place to keep the flag file in external storage if the function is stateless.

What to put in the push

• the failure rate in the window and the absolute numbers fail/total;
• the reason for the last failure (json.loads vs schema);
• a redacted sample of the raw response — it shows the fences / intro phrase;
• a hint about the source: a fresh model release or your deploy with a prompt edit.

Related recipes

• Schema-drift in tool calls — JSON valid, but fields shifted.
• Eval regression — same trigger: a model release.
• Toxic app output — another signal of broken output.