Build a production-grade structured extraction API: an OpenAI-compatible FastAPI service that extracts typed, validated data from unstructured text — with streaming, retry-on-validation-failure, per-tenant rate limiting, and a model fallback chain.
POST /v1/extract (OpenAI-compatible)
└─ validate_api_key(authorization)
└─ check_and_consume_quota(tenant_id, tier, estimated_tokens)
└─ rpm_limiter.acquire()
└─ extract_with_retry(text, schema, max_retries=3)
└─ Attempt 1: claude-haiku-4-5-20251001 + Pydantic validation
└─ Attempt 2 (if fail): add validation error to prompt, retry haiku
└─ Attempt 3 (if fail): escalate to claude-sonnet-4-6
└─ Fallback: rule_based_extractor()
└─ log_cost_record(tenant_id, model, tokens)
└─ return validated JSONProductInfo (name, price_usd, in_stock, tags), ArticleSummary (title, summary, topics, sentiment), and ContactInfo (name, email, phone, company). Use appropriate validators (ge=0 for price, enum for sentiment, EmailStr for email).get_extraction_tool(model_class) function that converts any Pydantic model to an Anthropic tool definition using model.model_json_schema(). Test it generates valid JSON schema for all three models.ValidationError is raised for all invalid inputs.extract_with_retry(text, model_class, max_retries=3) using instructor. On ValidationError, include the specific error message in the next prompt attempt.rule_based_fallback(text, model_class) that uses regex and heuristics to extract best-effort data for each schema. It should never raise — return defaults for unextractable fields.TokenBucket(capacity=10, rate=50/60) as a global rate limiter. Wrap all LLM calls with rpm_limiter.acquire(). Test by sending 20 requests in a burst — first 10 should return immediately, next 10 should be delayed.HTTP 429 with a Retry-After header when quota is exceeded.X-Quota-Used and X-Quota-Remaining response headers to every successful response so clients can implement proactive throttling.POST /v1/extract that accepts an OpenAI-compatible request body with an additional schema_type field ("product" | "article" | "contact") to select the Pydantic model.openai Python client with base_url pointing to your FastAPI server. Extract product info from 10 sample descriptions — verify all 10 return valid, schema-conformant JSON.GET /v1/quota/{tenant_id} endpoint that returns current token usage, daily limit, and resets_at timestamp. Test that it correctly reflects usage from the extraction endpoint.# Integration test — run end-to-end
from openai import OpenAI
client = OpenAI(
api_key="test-tenant-pro",
base_url="http://localhost:8000/v1",
)
SAMPLE_PRODUCTS = [
"Apple AirPods Pro 2nd gen, $249, in stock, noise-cancelling, wireless",
"Sony WH-1000XM5 headphones, currently unavailable, $349.99, premium audio",
"Anker SoundCore Q20i, $59.99, in stock, budget-friendly, Bluetooth",
]
for product_text in SAMPLE_PRODUCTS:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": product_text}],
extra_body={"schema_type": "product"}, # Custom extension field
)
import json
data = json.loads(response.choices[0].message.content)
print(f"Name: {data['name']}, Price: ${data['price_usd']}, In stock: {data['in_stock']}")
# → Name: Apple AirPods Pro 2nd gen, Price: $249.0, In stock: True
# → Name: Sony WH-1000XM5, Price: $349.99, In stock: False
# → Name: Anker SoundCore Q20i, Price: $59.99, In stock: TrueGraduation criteria: 100% of valid inputs produce schema-conformant output (via retry or fallback), rate limiter correctly delays burst requests, quota tracking returns accurate headers, and the endpoint works with a standard OpenAI client with zero modifications.