Browse documentation
FrameworksInstructor

Instructor

Integrate nRouter with Instructor for structured LLM outputs, automated retry logic, streaming responses, and strict Pydantic model validation in Python.

Last updated

Instructor supercharges Python language models with strict schema enforcement, structured data extraction, and automated retry cycles powered by Pydantic. By pointing Instructor at https://api.nrouter.ai/v1, your application gains enterprise-grade gateway controls: prompt injection defense, server-side guardrails, semantic caching, unified provider failover, and exact per-request cost attribution.

Because nRouter exposes standard OpenAI-compatible endpoints, Instructor wraps either the official openai Python client or the native nrouter-sdk seamlessly. Guardrails and content policies run at the gateway layer before the request reaches downstream model providers, preventing malformed inputs and toxic content from polluting structured output pipelines.

Prerequisites & Installation

Instructor requires Python 3.10 or higher. For production services, we recommend managing dependencies inside an isolated virtual environment (venv or uv).

Install Instructor along with Pydantic and the OpenAI client library:

pip install instructor openai pydantic

If you plan to use the official nRouter Python SDK for direct metadata inspection and typed client helpers, install nrouter-sdk:

pip install instructor nrouter-sdk pydantic

Setup & Configuration

Initialize your client by supplying your virtual API key and configuring the nRouter base URL. Virtual keys are provisioned in the nRouter dashboard and allow fine-grained controls, team budgets, and per-key rate limits.

export NROUTER_API_KEY="sk-nrouter-your-virtual-key"

Basic Client Setup

Using the standard openai library patched with Instructor:

import os
import instructor
from openai import OpenAI

# Initialize standard OpenAI client pointed at nRouter gateway
client = instructor.from_openai(
    OpenAI(
        base_url="https://api.nrouter.ai/v1",
        api_key=os.environ["NROUTER_API_KEY"],
        timeout=30.0,
        max_retries=3,
    )
)

Configuration Parameters

When configuring your client for production workloads, tune the following connection and retry parameters:

ParameterTypeDefaultDescription
base_urlstrhttps://api.nrouter.ai/v1nRouter unified gateway endpoint. Must include the /v1 suffix.
api_keystrNoneYour nRouter virtual key (sk-nrouter-...). Avoid hardcoding in source code.
timeoutfloat60.0Total request timeout in seconds, covering DNS resolution, TLS handshake, and token generation.
max_retriesint2Client-side HTTP retry attempts for network blips. nRouter handles upstream provider retries automatically.
default_headersdict{}Custom HTTP headers sent with every request, such as x-nr-routing or correlation IDs.
import os
import httpx
import instructor
from openai import OpenAI

# Advanced client configuration with custom HTTP transport
http_client = httpx.Client(
    timeout=httpx.Timeout(connect=5.0, read=45.0, write=10.0, pool=60.0),
    limits=httpx.Limits(max_keepalive_connections=50, max_connections=200),
)

client = instructor.from_openai(
    OpenAI(
        base_url="https://api.nrouter.ai/v1",
        api_key=os.environ["NROUTER_API_KEY"],
        http_client=http_client,
        max_retries=2,
        default_headers={
            "x-nr-routing": "latency",
        },
    )
)

Structured Extraction Patterns

Instructor leverages Pydantic models to define expected JSON schemas. When calling client.chat.completions.create, specify response_model with your Pydantic class.

Single Entity Extraction

Extract strongly typed entity attributes from unstructured natural language:

from pydantic import BaseModel, Field

class UserProfile(BaseModel):
    name: str = Field(description="Full name of the user")
    skills: list[str] = Field(description="List of verified technical skills")
    experience_level: str = Field(description="Junior, Mid, Senior, or Staff")
    years_of_experience: int = Field(description="Total years of professional engineering experience")

user_data = client.chat.completions.create(
    model="gpt-5.4-mini",
    response_model=UserProfile,
    max_completion_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "Jason has 8 years of experience building distributed systems in Rust and Go.",
        }
    ],
)

print(f"Name: {user_data.name}")
print(f"Level: {user_data.experience_level}")
print(f"Skills: {', '.join(user_data.skills)}")
print(f"Years: {user_data.years_of_experience}")

Nested Schemas & Validation Logic

Combine nested Pydantic models with field validators. If the model generates data that violates a Pydantic validator, Instructor automatically feeds the validation error back to the model for self-correction:

from typing import List, Optional
from pydantic import BaseModel, Field, field_validator

class SecurityFinding(BaseModel):
    cve_id: Optional[str] = Field(None, description="CVE identifier if applicable")
    severity: str = Field(description="Low, Medium, High, or Critical")
    description: str

    @field_validator("severity")
    def validate_severity(cls, v: str) -> str:
        allowed = {"Low", "Medium", "High", "Critical"}
        if v.capitalize() not in allowed:
            raise ValueError(f"Severity must be one of {allowed}")
        return v.capitalize()

class AuditReport(BaseModel):
    repository: str
    findings: List[SecurityFinding]
    compliance_passed: bool

report = client.chat.completions.create(
    model="gpt-5.4-mini",
    response_model=AuditReport,
    max_retries=3,  # Instructor auto-corrects validation failures up to 3 times
    messages=[
        {
            "role": "system",
            "content": "Analyze the following scan log and produce a structured audit report.",
        },
        {
            "role": "user",
            "content": "Repository: api-gateway. Found CVE-2026-10492 rated Critical in auth token parser.",
        },
    ],
)

print(f"Audit for {report.repository}: Passed = {report.compliance_passed}")
for item in report.findings:
    print(f" - [{item.severity}] {item.cve_id}: {item.description}")

Production Best Practices

Deterministic Routing & Fallbacks

Configure routing strategies at the request level using the x-nr-routing header or by specifying fallback chains:

# Pass custom headers per request via extra_headers
response = client.chat.completions.create(
    model="gpt-5.4-mini,claude-haiku-4-5-20251001",
    response_model=UserProfile,
    messages=[{"role": "user", "content": "Extract profile for Sarah Connor"}],
    extra_headers={
        "x-nr-routing": "latency",
    },
)
  • x-nr-routing: latency: Directs traffic to the lowest-latency active provider deployment.
  • x-nr-routing: cost: Prioritizes the most cost-effective deployment matching the model specification.
  • Model Fallback List: When providing a comma-delimited model list (model1,model2), nRouter attempts model1 and immediately falls back to model2 if the primary provider encounters capacity constraints.

Semantic Caching

Enable semantic caching on repeated structured extractions to reduce latency and eliminate redundant LLM inference costs. Cache rules can be set globally in the nRouter dashboard or overridden per request:

response = client.chat.completions.create(
    model="gpt-5.4-mini",
    response_model=UserProfile,
    messages=[{"role": "user", "content": "Extract profile for John Doe"}],
    extra_body={
        "nrouter_cache": True,
    },
)

Telemetry & FinOps Tracking

Every call through nRouter returns transparent telemetry headers:

  • x-nr-request-id: Unique UUID assigned at the edge gateway, bound to request traces and audit logs.
  • x-nr-model: Canonical identifier of the model deployment that fulfilled the extraction.
  • x-nr-cost-status: Either exact (fully calculated) or unpriced.
  • x-nr-request-cost: Exact USD cost charged for the request based on provider token pricing.
  • x-nr-input-tokens / x-nr-output-tokens: Precise token accounting figures.

When using the native nrouter-sdk, these values are automatically parsed and exposed on the client instance:

from nroutersdk import nRouter
import instructor

raw_client = nRouter()
instructor_client = instructor.from_openai(raw_client)

data = instructor_client.chat.completions.create(
    model="gpt-5.4-mini",
    response_model=UserProfile,
    messages=[{"role": "user", "content": "Extract profile for Alice"}],
)

if raw_client.last_response:
    meta = raw_client.last_response
    print(f"Request ID: {meta.request_id}")
    print(f"Cost: ${meta.cost} ({meta.cost_status})")
    print(f"Tokens: {meta.input_tokens} prompt / {meta.output_tokens} completion")

Troubleshooting & Error Handling

nRouter maps errors to clear HTTP status codes. When Instructor encounters a non-200 response, it bubbles up standard OpenAI API exceptions.

Common Error Codes

StatusCodeCauseRecommended Action
400guardrail_blockedInput violated content security policy or prompt injection filterReview input prompt; verify safety guidelines in nRouter dashboard.
400invalid_request_errorMalformed JSON schema or unsupported model parametersVerify Pydantic schema structure and token limits.
401authentication_errorMissing, malformed, or revoked NROUTER_API_KEYCheck environment variable; mint a fresh virtual key in the dashboard.
402insufficient_creditsOrganization balance is zero or key budget ceiling reachedTop up balance or adjust virtual key spending limits.
429rate_limit_exceededRequests per minute (RPM) or tokens per minute (TPM) limit reachedImplement exponential backoff; check retry-after header.
500 / 503service_unavailableDownstream provider outage or temporary gateway disruptionEnsure fallback models are configured; nRouter routes around provider outages.

Error Recovery Example

import openai
from pydantic import BaseModel

class ProductInfo(BaseModel):
    title: str
    price: float

try:
    product = client.chat.completions.create(
        model="gpt-5.4-mini",
        response_model=ProductInfo,
        messages=[{"role": "user", "content": "Product: Wireless Mouse for $24.99"}],
    )
    print(product)
except openai.AuthenticationError:
    print("Authentication failed: Verify NROUTER_API_KEY.")
except openai.RateLimitError as e:
    print(f"Rate limited by nRouter: {e}. Backing off.")
except openai.BadRequestError as e:
    if "guardrail" in str(e).lower():
        print("Request was rejected by nRouter safety guardrails.")
    else:
        print(f"Invalid request parameters: {e}")
except openai.APIError as e:
    print(f"Gateway error ({e.code}): {e.message}")

Next Steps

Was this page helpful?