Metrics tracked in-process (use a metrics store for production)

title: "How to Monitor Instructor with Vigilmon" published: true description: Instructor is a Python library for structured, type-safe LLM outputs using Pydantic validation. Here's how to monitor Instructor-powered extraction pipelines, track validation failure rates, and alert when your structured output workflows degrade. tags: instructor, pydantic, llm, monitoring cover_image:

Instructor patches OpenAI, Anthropic, Gemini, and Cohere clients to guarantee JSON schema compliance via automatic retry-with-correction on Pydantic validation failures. It turns unreliable LLM outputs into structured data your downstream code can trust. But when the LLM changes its output format, when validation enters a retry loop, or when a batch extraction job stalls on a bad document, your pipelines silently degrade — retries eat latency budget, costs spike, and downstream consumers receive incomplete data. Vigilmon gives you heartbeat monitoring for extraction jobs, HTTP health checks for inference services, and instant alerts so you catch Instructor failures before they cascade.

What You'll Set Up

• Heartbeat monitor for scheduled extraction and classification jobs
• HTTP health endpoint that reports validation failure rates
• Alert channels for on-call notification

Prerequisites

• An application using Instructor with OpenAI, Anthropic, or another supported provider
• A free Vigilmon account

Step 1: Add a Health Endpoint to Your Instructor Service

Instructor runs inside your application process. Add a /health endpoint that reports the current state of your extraction service and tracks validation metrics:

import os
import time
import instructor
import openai
from pydantic import BaseModel
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

_metrics = {
    "total_calls": 0,
    "validation_failures": 0,
    "retries": 0,
    "last_success_ts": None,
}


class HealthStatus(BaseModel):
    status: str
    total_calls: int
    validation_failures: int
    retries: int
    failure_rate_pct: float


@app.get("/health")
async def health():
    """Fast health check — reports extraction service metrics."""
    total = _metrics["total_calls"]
    failures = _metrics["validation_failures"]
    failure_rate = round((failures / total * 100) if total > 0 else 0.0, 2)

    # Alert if failure rate exceeds 20%
    status = "ok" if failure_rate < 20.0 else "degraded"

    return JSONResponse(
        status_code=200 if status == "ok" else 503,
        content={
            "status": status,
            "total_calls": total,
            "validation_failures": failures,
            "retries": _metrics["retries"],
            "failure_rate_pct": failure_rate,
        },
    )


@app.get("/health/deep")
async def deep_health():
    """Deep health check — performs a real Instructor extraction."""
    class ProbeResponse(BaseModel):
        message: str

    client = instructor.from_openai(openai.AsyncOpenAI())
    start = time.monotonic()

    try:
        result = await client.chat.completions.create(
            model="gpt-4o-mini",
            response_model=ProbeResponse,
            messages=[{"role": "user", "content": 'Reply with JSON: {"message": "ok"}'}],
            max_retries=1,
        )
        latency_ms = int((time.monotonic() - start) * 1000)
        return {
            "status": "ok",
            "message": result.message,
            "latency_ms": latency_ms,
        }
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "detail": str(e)},
        )

The /health endpoint exposes live validation failure rates — when your LLM starts producing malformed outputs that fail Pydantic validation, this metric rises and Vigilmon alerts you.

Step 2: Instrument Instructor Calls to Track Failures

Wrap your Instructor client to capture metrics on every call:

import instructor
import openai
import anthropic
from pydantic import BaseModel, ValidationError
from typing import Type, TypeVar

T = TypeVar("T", bound=BaseModel)

openai_client = instructor.from_openai(openai.AsyncOpenAI())


async def extract(
    response_model: Type[T],
    messages: list[dict],
    model: str = "gpt-4o-mini",
    max_retries: int = 3,
) -> T:
    """Instructor extraction wrapper with metrics tracking."""
    _metrics["total_calls"] += 1

    retry_count = 0

    # Instructor handles retries internally; hook into its retry callback
    async def on_retry(response, exception, **kwargs):
        nonlocal retry_count
        retry_count += 1
        _metrics["retries"] += 1

    try:
        result = await openai_client.chat.completions.create(
            model=model,
            response_model=response_model,
            messages=messages,
            max_retries=max_retries,
            hooks={"on_error": on_retry},  # Instructor >=1.0 hook API
        )
        _metrics["last_success_ts"] = time.time()
        return result

    except instructor.exceptions.InstructorRetryException as e:
        # Exhausted all retries — validation never passed
        _metrics["validation_failures"] += 1
        raise

    except Exception:
        _metrics["validation_failures"] += 1
        raise

Note: The hooks API is available in Instructor >= 1.0. For earlier versions, wrap the call in a try/except on ValidationError or instructor.exceptions.InstructorRetryException.

Step 3: Heartbeat Monitoring for Batch Extraction Jobs

Batch extraction jobs — nightly document classification, daily entity extraction runs, weekly report generation — have no persistent HTTP listener. HTTP monitoring cannot catch silent failures. Use a Vigilmon heartbeat:

1. In Vigilmon, click Add Monitor → Cron Heartbeat.
2. Set the expected ping interval to match your job schedule (e.g. 1440 minutes for a daily job).
3. Copy the heartbeat URL.

Instrument your batch job to ping the heartbeat after each successful run:

import os
import httpx
import logging
from pathlib import Path

logger = logging.getLogger(__name__)

VIGILMON_HEARTBEAT_URL = os.environ.get("VIGILMON_HEARTBEAT_URL")


class DocumentClassification(BaseModel):
    category: str
    confidence: float
    summary: str


async def classify_document(text: str) -> DocumentClassification:
    return await extract(
        response_model=DocumentClassification,
        messages=[
            {
                "role": "system",
                "content": "Classify the document into a category and provide a brief summary.",
            },
            {"role": "user", "content": text},
        ],
    )


async def run_nightly_classification():
    """Classify all pending documents and ping heartbeat on success."""
    pending_dir = Path(os.environ["PENDING_DOCS_DIR"])
    failed = []

    for doc_path in pending_dir.glob("*.txt"):
        try:
            text = doc_path.read_text(encoding="utf-8")
            classification = await classify_document(text)
            await save_classification(doc_path.stem, classification)
            doc_path.unlink()
            logger.info(f"Classified {doc_path.name}: {classification.category}")
        except Exception as e:
            logger.error(f"Failed to classify {doc_path.name}: {e}")
            failed.append(doc_path.name)

    if failed:
        logger.error(f"Classification job completed with failures: {failed}")
        # Do NOT ping heartbeat — absence of ping is the alert
        return

    if VIGILMON_HEARTBEAT_URL:
        async with httpx.AsyncClient() as client:
            await client.get(VIGILMON_HEARTBEAT_URL, timeout=10)
        logger.info("Heartbeat pinged — classification job succeeded")


async def save_classification(doc_id: str, result: DocumentClassification):
    """Stub: persist classification result."""
    pass

If the job fails because the LLM produces invalid JSON that exhausts all retries, or because a document causes a downstream error, no heartbeat is sent and Vigilmon alerts you within one missed interval.

Step 4: Set Up HTTP Monitoring for the Inference Service

With /health and /health/deep deployed, add Vigilmon monitors:

1. Log in to vigilmon.online and click Add Monitor.
2. Set Type to HTTP / HTTPS.
3. URL: https://your-service.com/health
4. Check interval: 5 minutes
5. Expected HTTP status: 200
6. Keyword check: "ok" to confirm failure rate is within threshold
7. Click Save.

Add the deep health monitor for end-to-end extraction validation:

1. Click Add Monitor → HTTP / HTTPS.
2. URL: https://your-service.com/health/deep
3. Check interval: 15 minutes — this makes a real LLM call, which costs tokens
4. Response timeout: 30 seconds
5. Keyword check: "ok"
6. Click Save.

Step 5: Alert on Validation Failure Spikes

Configure alert thresholds in Vigilmon for your /health endpoint. The endpoint returns HTTP 503 when validation failure rate exceeds 20%. But you may also want early warnings:

Extend the health endpoint to return a degraded warning status at lower thresholds:

@app.get("/health")
async def health():
    total = _metrics["total_calls"]
    failures = _metrics["validation_failures"]
    failure_rate = round((failures / total * 100) if total > 0 else 0.0, 2)

    if failure_rate >= 20.0:
        status = "critical"
        http_status = 503
    elif failure_rate >= 10.0:
        status = "warning"
        http_status = 200  # Still 200 so Vigilmon doesn't page on-call
    else:
        status = "ok"
        http_status = 200

    return JSONResponse(
        status_code=http_status,
        content={
            "status": status,
            "failure_rate_pct": failure_rate,
            "total_calls": total,
            "retries": _metrics["retries"],
        },
    )

Set up a separate Vigilmon monitor with a keyword check for "warning" to catch early degradation before it reaches the critical threshold — route it to a lower-priority channel (email) while routing the 503 check to your Slack on-call channel.

Step 6: Configure Alert Channels

1. Go to Alert Channels in Vigilmon and add Slack, email, or a webhook.
2. For the heartbeat monitor, keep the default 1 missed interval — a missed batch job means your structured data pipeline has stopped.
3. For the HTTP deep health monitor, set Consecutive failures before alert to 2 — the LLM occasionally returns malformed output that self-corrects on the next check interval.

Use the Vigilmon API to suppress alerts during planned model migrations when higher validation failure rates are expected:

curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{"monitor_id": "abc123", "duration_minutes": 30}'

# Switch to new model and observe validation rates
python migrate_to_new_model.py

Summary

| Monitor | Target | What It Catches | |---|---|---| | Heartbeat | Heartbeat URL | Batch extraction job crash, missed schedule | | HTTP health | /health | Rising validation failure rate, service crash | | Deep health | /health/deep | End-to-end extraction failure, LLM unreachable | | Warning check | /health keyword "warning" | Early validation degradation before critical |

Instructor gives your application reliable structured outputs from LLMs — Vigilmon ensures the pipelines that depend on those outputs are monitored continuously. With heartbeats on every batch job and health checks that track validation failure rates, you'll catch schema compliance degradation before it silently corrupts your downstream data.

Get started free at vigilmon.online — your first monitor is running in under a minute.