Instructor library structured outputs through a gateway

The Instructor library removes the drudgery of parsing LLM output by mapping responses directly to Pydantic models. Standing up an Instructor library structured outputs gateway means you get that type safety while routing requests through a single OpenAI-compatible endpoint that can reach many providers. This how-to takes you from an empty directory to a running extraction pipeline with fallback, caching hints, and per-token metering.

Step 1: Install the toolchain

Create a fresh environment and pull the dependencies:

python -m venv .venv
source .venv/bin/activate
pip install instructor openai pydantic python-dotenv

Instructor wraps the official OpenAI SDK, so you keep the same interface you already know. Pydantic handles schema definition and validation. Keep both libraries current; Instructor tracks OpenAI SDK changes closely and breaks on major version shifts.

Step 2: Define your output contract

Suppose you need to extract invoice line items from unstructured text. Declare a Pydantic model that describes the exact shape you expect:

from pydantic import BaseModel, Field
from typing import List

class LineItem(BaseModel):
    quantity: int = Field(..., gt=0)
    unit_price_cents: int = Field(..., ge=0)

class Invoice(BaseModel):
    vendor: str
    items: List[LineItem]
    total_cents: int

Why field descriptions matter

The descriptions are not cosmetic. Instructor forwards them to the model as part of the function/tool schema, which measurably improves extraction accuracy. Use strict types: integers for money avoid floating-point rounding bugs, and gt/ge constraints reject garbage before it hits your business logic.

Step 3: Point the client at your gateway

Instructor works by patching the OpenAI client. You only change base_url and api_key. A gateway gives you one URL for many models.

import os
from openai import OpenAI
import instructor

client = instructor.from_openai(
    OpenAI(
        base_url="https://api.n4n.ai/v1",  # OpenAI-compatible, 240+ models
        api_key=os.environ["GATEWAY_KEY"],
    )
)

If you use a different gateway, swap the base_url. The key point: the Instructor library structured outputs gateway pattern decouples your schema logic from the underlying model provider. n4n.ai forwards provider cache-control hints and honors client routing directives, so you can pin a model or let the gateway choose a healthy backend.

Step 4: Execute a structured extraction

Call the client with response_model instead of hand-parsing messages:

text = """
Acme Corp
2x Widget @ 500 cents
1x Sprocket @ 1200 cents
"""

invoice = client.chat.completions.create(
    model="anthropic/claude-3.5-sonnet",
    response_model=Invoice,
    messages=[{"role": "user", "content": f"Extract invoice: {text}"}],
)

print(invoice.model_dump_json(indent=2))

The returned object is a validated Invoice. If the model emits malformed JSON, Instructor retries with correction feedback automatically. That retry loop is the main reason to use the library rather than raw json.loads.

Step 5: Route and control caching through the gateway

Gateways accept headers that providers understand. To force a specific provider or enable prompt caching, pass extra_headers:

invoice = client.chat.completions.create(
    model="openai/gpt-4o-mini",
    response_model=Invoice,
    messages=[{"role": "user", "content": f"Extract invoice: {text}"}],
    extra_headers={
        "x-router-prefer": "openai",
        "cache-control": "max-age=3600"
    },
)

The gateway honors these hints. If the preferred provider is rate-limited, automatic fallback kicks in and you still get a response. Per-token usage metering on the gateway side lets you attribute cost without instrumenting your own middleware.

Inspecting the raw request

For debugging, log the outgoing payload:

import json
req = client.chat.completions._client._build_request(
    method="POST",
    url="chat/completions",
    json={"model": "openai/gpt-4o-mini", "messages": []}
)
print(req.headers)

This shows whether your cache-control header survived serialization.

Step 6: Verify the pipeline works

Success means three things: the call returns without raising, the object validates, and the numbers add up. Add a quick assertion script:

def verify(inv: Invoice):
    assert inv.vendor, "Missing vendor"
    computed = sum(i.quantity * i.unit_price_cents for i in inv.items)
    assert computed == inv.total_cents, "Total mismatch"
    print("OK", inv.total_cents)

verify(invoice)

Run it. If you see OK and the JSON, the Instructor library structured outputs gateway integration is functional. For CI, wrap this in pytest and mock the gateway with respx if you want offline tests.

Step 7: Compare with direct provider coupling

Using Instructor against OpenAI directly looks identical except base_url is omitted and model is a single string like "gpt-4o". The difference surfaces in production:

That trade-off is why an Instructor library structured outputs gateway is superior for multi-provider systems, despite the tiny extra base_url config.

Step 8: Stream structured outputs

For long documents, streaming avoids head-of-line latency. Instructor supports stream=True with iterable responses:

import asyncio
from openai import OpenAI

async def stream_invoice():
    async_client = instructor.from_openai(
        OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["GATEWAY_KEY"])
    )
    async for inv in await async_client.chat.completions.create(
        model="anthropic/claude-3.5-sonnet",
        response_model=Invoice,
        messages=[{"role": "user", "content": "Extract from large PDF text..."}],
        stream=True,
    ):
        print("Partial:", inv.model_dump())

asyncio.run(stream_invoice())

Note: not all gateways support streaming with structured outputs equally; verify with a short doc before wiring a 200-page pipeline.

Step 9: Handle errors and partial schemas

Instructor raises ValidationError from Pydantic on unrecoverable mismatch. Catch it and fall back to a stronger model:

from pydantic import ValidationError

def extract_with_fallback(text: str) -> Invoice:
    try:
        return client.chat.completions.create(
            model="meta-llama/llama-3.1-8b",
            response_model=Invoice,
            messages=[{"role": "user", "content": text}],
        )
    except ValidationError:
        return client.chat.completions.create(
            model="anthropic/claude-3.5-sonnet",
            response_model=Invoice,
            messages=[{"role": "user", "content": text}],
        )

This pattern works because the gateway abstracts model names. You are not rewriting prompt logic, only changing the model string.

Step 10: Productionize with environment config

Never hardcode keys. Use .env:

GATEWAY_KEY=sk-...
GATEWAY_BASE=https://api.n4n.ai/v1

from dotenv import load_dotenv
load_dotenv()

client = instructor.from_openai(
    OpenAI(
        base_url=os.environ["GATEWAY_BASE"],
        api_key=os.environ["GATEWAY_KEY"],
        timeout=30,
    )
)

Wrap the client in a factory function. Add timeouts as shown. Monitor gateway usage metrics from its dashboard to spot drift in token spend or latency.

Step 11: Extend to batch and async workers

Once the single-call path is verified, push throughput with concurrent requests:

import asyncio
from typing import Iterable

async def batch_extract(texts: Iterable[str]) -> list[Invoice]:
    async_client = instructor.from_openai(
        OpenAI(base_url=os.environ["GATEWAY_BASE"], api_key=os.environ["GATEWAY_KEY"])
    )
    tasks = [
        async_client.chat.completions.create(
            model="openai/gpt-4o-mini",
            response_model=Invoice,
            messages=[{"role": "user", "content": t}],
        )
        for t in texts
    ]
    return await asyncio.gather(*tasks)

The gateway’s fallback and metering apply per call, so batch failures degrade gracefully instead of taking down the whole job.

Closing notes

The Instructor library structured outputs gateway approach is fundamentally about separation of concerns: schemas live in your code, model selection lives in the gateway. You get typed outputs, provider redundancy, and consolidated metering without bespoke adapter layers. Start with the steps above, then expand to batch extraction and async workers as load grows.