DEV Community

Morgan Xu
Morgan Xu

Posted on

Before You Plug a Model Into Your Pipeline, Make It Pass a Contract Test

Why this is worth reading

A new model release often claims strong JSON or tool-calling behavior, but your task fails because the output is shaped wrong, not because the model lacks "intelligence". You will learn to define a small output contract, run it against any OpenAI-compatible endpoint with a repeatable Python script, and interpret the results without trusting a leaderboard. This is a smoke test for the exact failure that breaks pipelines: drift in structure and assertions. It runs cheaply, and the free-infrastructure section below mentions one availability option.

The failure mode most evals do not catch

Most benchmark numbers summarize average performance across someone else's distribution. They do not tell you whether a model will emit the correct JSON shape for your incident summarizer, keep duplicate action IDs out, or reject extra keys that your types ignore. When that contract breaks, you notice as a runtime error after deployment, usually as NoneType or a schema validation exception in code you did not write. The fix is to stop comparing models as interchangeable generators and instead treat each one as a candidate implementation of a contract.

A contract has two parts: a JSON Schema and a small set of task-specific assertions. The schema checks shape; the assertions check meaning and invariants. Neither is enough alone. A model can return valid JSON that says nothing useful; or it can return useful text that crashes your parser. The harness below makes both checks repeatable.

Start with the contract, not the prompt

Create contracts/incident.json:

{
  "schema": {
    "type": "object",
    "required": ["summary", "actions", "confidence"],
    "additionalProperties": false,
    "properties": {
      "summary": {"type": "string", "minLength": 10},
      "actions": {
        "type": "array",
        "items": {
          "type": "object",
          "required": ["id", "command"],
          "additionalProperties": false,
          "properties": {
            "id": {"type": "string"},
            "command": {"type": "string"}
          }
        }
      },
      "confidence": {"type": "number", "minimum": 0.0, "maximum": 1.0}
    }
  }
}
Enter fullscreen mode Exit fullscreen mode

Then define assertions in contracts/incident_checks.py:

def summary_is_not_a_refusal(doc):
    return doc['summary'].strip().lower() not in {
        'as an ai language model',
        'i cannot help',
        'i am unable to'
    }


def action_ids_are_unique(doc):
    ids = [a['id'] for a in doc['actions']]
    return len(ids) == len(set(ids))


def confidence_is_usable(doc):
    return doc['confidence'] >= 0.0 and doc['confidence'] <= 1.0
Enter fullscreen mode Exit fullscreen mode

Notice you are not asking the model to "try harder" in the prompt. You are defining the failure condition first, which makes a pass less ambiguous.

A minimal contract probe you can run anywhere

The script accepts an endpoint, API key, model name, and input file. It sends a single request, parses the first JSON object in the completion, validates schema, runs assertions, and prints a compact result plus the first failure. It is deliberately small so you can read every line before sending your data anywhere.

Save as contract_probe.py:

#!/usr/bin/env python3
import argparse
import json
import os
import sys
import time
import requests
from jsonschema import Draft202012Validator


def first_json_object(text):
    decoder = json.JSONDecoder()
    for i, ch in enumerate(text):
        if ch == '{':
            try:
                obj, _ = decoder.raw_decode(text[i:])
                return obj
            except json.JSONDecodeError:
                continue
    raise ValueError('no JSON object found')


def main():
    p = argparse.ArgumentParser()
    p.add_argument('--endpoint', default=os.environ.get('MODEL_ENDPOINT'))
    p.add_argument('--key', default=os.environ.get('MODEL_KEY', 'not-used'))
    p.add_argument('--model', default=os.environ.get('MODEL_NAME'))
    p.add_argument('--prompt', required=True)
    p.add_argument('--contract', required=True)
    p.add_argument('--temperature', type=float, default=0.0)
    args = p.parse_args()

    contract = json.load(open(args.contract))
    schema = contract['schema']
    prompt = open(args.prompt).read()

    checks = {}
    checks_file = args.contract.replace('.json', '_checks.py')
    if os.path.exists(checks_file):
        import importlib.util
        spec = importlib.util.spec_from_file_location('checks', checks_file)
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        checks = {
            name: getattr(mod, name)
            for name in dir(mod)
            if callable(getattr(mod, name)) and not name.startswith('_')
        }

    start = time.time()
    if not args.endpoint or not args.model:
        print('MODEL_ENDPOINT and MODEL_NAME are required')
        sys.exit(2)

    payload = {
        'model': args.model,
        'temperature': args.temperature,
        'messages': [
            {
                'role': 'system',
                'content': 'Return exactly one JSON object matching the provided requirement and no prose.',
            },
            {
                'role': 'user',
                'content': prompt,
            },
        ],
    }

    try:
        r = requests.post(
            args.endpoint,
            headers={'Authorization': f'Bearer {args.key}'},
            json=payload,
            timeout=60,
        )
        r.raise_for_status()
        content = r.json()['choices'][0]['message']['content']
        doc = first_json_object(content)
        elapsed = time.time() - start

        validator = Draft202012Validator(schema)
        schema_errors = sorted(e.message for e in validator.iter_errors(doc))
        assertion_failures = [
            name for name, fn in checks.items() if not fn(doc)
        ]
        outcome = 'PASS' if not schema_errors and not assertion_failures else 'FAIL'
        print(json.dumps({
            'outcome': outcome,
            'latency_s': round(elapsed, 2),
            'schema_errors': schema_errors[:3],
            'assertion_failures': assertion_failures,
            'token_or_choice_count': len(content),
        }, indent=2))
    except Exception as e:
        print(json.dumps({
            'outcome': 'ERROR',
            'error_type': type(e).__name__,
            'error': str(e)[:400],
            'latency_s': round(time.time() - start, 2),
        }, indent=2))


if __name__ == '__main__':
    main()
Enter fullscreen mode Exit fullscreen mode

Run it locally:

python -m venv .venv && source .venv/bin/activate
pip install requests jsonschema
mkdir -p contracts
# add contracts/incident.json and contracts/incident_checks.py from above
cat > incident.txt <<'EOF'
A cron job on db-3 failed with OOM at 04:11. It was restarted once,
the second run completed at 04:26, and no rows were lost.
EOF

python contract_probe.py --endpoint "$MODEL_ENDPOINT" --model "$MODEL_NAME" --prompt incident.txt --contract contracts/incident.json
Enter fullscreen mode Exit fullscreen mode

The result is a JSON line, so you can redirect to a file and diff across repeated runs. Repeat the probe three to five times because a single lucky parse is not evidence. Count a model as passable only when every run returns PASS and the failure mode is stable when it does not.

Interpret failures without guessing

Use this table when a run falls over, because the failure category tells you where to spend time:

If you see What it usually means What to do next
schema_errors includes additionalProperties The model returns extra fields you did not ask for Tighten additionalProperties: false and add those fields only if they should be in the contract
schema_errors on confidence Number emitted as string or outside 0-1 Normalize or reject at the model boundary; do not silently patch downstream
summary_is_not_a_refusal Prompt/safety refusal without useful output Treat as a capability miss for this task, not a reason to lower the threshold
action_ids_are_unique Duplicate identifiers Your contract needs a uniqueness invariant; schema cannot express it
ERROR with HTTPError Endpoint, auth, or quota problem Check the dashboard limits and retry with a cool-off, not in a tight loop

The useful conclusion from a probe is not "model A is better than model B." It is "model A keeps violating the additionalProperties rule, so I can either fix my prompt, add a repair step, or exclude it from this extraction path."

Where a free endpoint and free server fit

You do not need to send every probe to an expensive hosted account. The script above is small enough to run from a free CI runner or a low-cost server. One option the operator of this content is making available is MonkeyCode, which advertises free model access and a free server option. Disclosure: This article was prepared as part of MonkeyCode's product outreach. Verify current quotas, model names, endpoint shape, and retention rules in the dashboard before you commit any data; the contract harness itself does not depend on MonkeyCode and works with any OpenAI-compatible endpoint.

For this kind of smoke test, use the free tier only for non-sensitive prompts and public fixtures. Do not log API keys, and do not treat a free endpoint as an availability guarantee for production.

Limitations you should be honest about

The contract harness is not an eval suite. JSON Schema can confirm shape and a few invariants, but it cannot tell you whether the summary is true or whether the command is safe. It also cannot measure factuality, tone, latency under load, privacy, or cost. If a model passes the probe, you have cleared only the cheapest bar: the output is parseable and obeys the contract you wrote. You still need golden examples, a human review for risky commands, and production monitoring.

The artifacts are a plan, not a claim that any specific model or free tier will behave the same for your traffic or stay free tomorrow. Product limits change, and a public model endpoint may rotate models or alter routing without notice.

Who should skip this

Skip or adapt this approach if your output is long-form prose, creative copy, or open-ended advice: a JSON schema adds clutter without improving quality. Skip it also if you are sending sensitive customer data to an external endpoint; run a local model or a private deployment instead. And skip it if you are trying to replace a full acceptance suite: this is a gate, not a green light.

Put it in front of the next model you try

Write the contract before you read the model card. Run the probe from a free server, repeat it enough to separate bad luck from bad outputs, and keep the failures where you can see them. The model is not ready until your parser can trust the shape, and the most useful model evaluation you can run today is the one you write around the exact JSON your pipeline expects.

If you try it, the comment worth leaving is which assertion caught the first real break: shape, uniqueness, or number type. That tells the next reader more than any leaderboard table will.

Top comments (0)