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 pydanticIf 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 pydanticSetup & 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:
| Parameter | Type | Default | Description |
|---|---|---|---|
base_url | str | https://api.nrouter.ai/v1 | nRouter unified gateway endpoint. Must include the /v1 suffix. |
api_key | str | None | Your nRouter virtual key (sk-nrouter-...). Avoid hardcoding in source code. |
timeout | float | 60.0 | Total request timeout in seconds, covering DNS resolution, TLS handshake, and token generation. |
max_retries | int | 2 | Client-side HTTP retry attempts for network blips. nRouter handles upstream provider retries automatically. |
default_headers | dict | {} | 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: Eitherexact(fully calculated) orunpriced.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
| Status | Code | Cause | Recommended Action |
|---|---|---|---|
400 | guardrail_blocked | Input violated content security policy or prompt injection filter | Review input prompt; verify safety guidelines in nRouter dashboard. |
400 | invalid_request_error | Malformed JSON schema or unsupported model parameters | Verify Pydantic schema structure and token limits. |
401 | authentication_error | Missing, malformed, or revoked NROUTER_API_KEY | Check environment variable; mint a fresh virtual key in the dashboard. |
402 | insufficient_credits | Organization balance is zero or key budget ceiling reached | Top up balance or adjust virtual key spending limits. |
429 | rate_limit_exceeded | Requests per minute (RPM) or tokens per minute (TPM) limit reached | Implement exponential backoff; check retry-after header. |
500 / 503 | service_unavailable | Downstream provider outage or temporary gateway disruption | Ensure 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
- Python SDK Guide — In-depth guide to the official nRouter Python SDK
- LangChain Integration — Orchestrating chains and RAG workflows
- Chat Completions API — Complete HTTP wire specifications