Adding AI features to a Python backend sounds straightforward until you're live with thousands of users and your LLM bill spikes unexpectedly. We've learned that calling LLM APIs well involves more than just importing a library and making requests—it requires thoughtful SDKs, streaming for UX, structured output validation, and deliberate cost management.
Use the official SDK — Anthropic and OpenAI provide typed Python clients
The difference between third-party wrappers and official SDKs becomes clear fast. Official clients from OpenAI and Anthropic handle authentication, retries, type hints, and API changes gracefully. We've found that starting with these saves debugging time that you'd otherwise spend on HTTP headers and JSON serialization edge cases.
For OpenAI, install the official package and authenticate via environment variable:
pip install openaiThen use it directly in your code:
from openai import OpenAI
client = OpenAI() # reads OPENAI_API_KEY from env
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": "Summarize this invoice in one sentence."
}
],
temperature=0.7,
max_tokens=150
)
print(response.choices[0].message.content)The same pattern applies to Anthropic with their official SDK:
pip install anthropicfrom anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Extract the total amount from this receipt."
}
]
)
print(response.content[0].text)Both SDKs give you type hints, proper error handling, and automatic retry logic. They also document breaking changes proactively, so your code doesn't silently break when the API evolves.
Stream responses — Stream tokens to the frontend for better UX
Streaming transforms the user experience. Instead of waiting 3-5 seconds for the entire response, users see tokens arrive in real time. This is especially valuable for long-form outputs like report generation or code explanation—users feel the response is working immediately.
With OpenAI, use stream=True:
from openai import OpenAI
client = OpenAI()
def generate_report_stream(user_request: str):
with client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "You are a detailed business analyst."
},
{
"role": "user",
"content": user_request
}
],
stream=True,
temperature=0.8,
max_tokens=2000
) as stream:
for text in stream.text_stream:
yield textIn a FastAPI endpoint, yield these tokens directly to the client:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/api/report")
async def generate_report(request: dict):
return StreamingResponse(
generate_report_stream(request["query"]),
media_type="text/event-stream"
)On the frontend (React or Next.js), consume this as a ReadableStream:
# Pseudocode for frontend consumption
const response = await fetch('/api/report', {
method: 'POST',
body: JSON.stringify({ query: userInput })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
setOutput(prev => prev + decoder.decode(value));
}With Anthropic, streaming works similarly via iteration:
from anthropic import Anthropic
client = Anthropic()
def stream_analysis(text: str):
with client.messages.stream(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[
{
"role": "user",
"content": f"Analyze: {text}"
}
]
) as stream:
for text_chunk in stream.text_stream:
yield text_chunkStreaming also reduces perceived latency on slower connections and gives users a chance to stop the request early if needed.
Validate output — Parse structured output with Pydantic models
LLMs are probabilistic; they hallucinate, they misformat, they occasionally ignore instructions. Validating and parsing their output with Pydantic ensures your downstream code doesn't break on malformed JSON or missing fields.
Define a Pydantic model for the structure you expect:
from pydantic import BaseModel, Field
from typing import List
class InvoiceLineItem(BaseModel):
description: str
quantity: int
unit_price: float
total: float
class ExtractedInvoice(BaseModel):
vendor: str = Field(..., description="Name of the vendor")
invoice_date: str = Field(..., description="Date in YYYY-MM-DD format")
total_amount: float
line_items: List[InvoiceLineItem]
tax_amount: float | None = NoneUse OpenAI's response format specification to request JSON mode:
import json
from openai import OpenAI
client = OpenAI()
def extract_invoice(invoice_text: str) -> ExtractedInvoice:
response = client.chat.completions.create(
model="gpt-4o-mini",
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": """You extract structured data from invoice text.
Always return valid JSON matching this schema:
{
"vendor": "string",
"invoice_date": "YYYY-MM-DD",
"total_amount": number,
"line_items": [
{"description": "string", "quantity": number, "unit_price": number, "total": number}
],
"tax_amount": number or null
}"""
},
{
"role": "user",
"content": f"Extract invoice data: {invoice_text}"
}
],
temperature=0
)
raw_json = json.loads(response.choices[0].message.content)
return ExtractedInvoice(**raw_json)If parsing fails, catch the validation error and log it for debugging:
from pydantic import ValidationError
try:
invoice = extract_invoice(ocr_text)
except ValidationError as e:
logger.error(f"Invoice parsing failed: {e}")
# Re-prompt with stricter instructions or fallback
invoice = extract_invoice_with_retry(ocr_text)Anthropic also supports JSON output via the thinking block and explicit instructions:
from anthropic import Anthropic
import json
client = Anthropic()
def extract_with_claude(invoice_text: str) -> ExtractedInvoice:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[
{
"role": "user",
"content": f"""Extract invoice data and return ONLY valid JSON:
{json.dumps(ExtractedInvoice.model_json_schema(), indent=2)}
Invoice text:
{invoice_text}"""
}
]
)
raw_json = json.loads(response.content[0].text)
return ExtractedInvoice(**raw_json)Structured output prevents downstream errors and makes debugging easier when the LLM does miss the mark.
Control costs — Cache results and set token limits
LLM APIs charge per token. In production, costs scale quickly without guardrails. We manage this through response caching, request deduplication, and token budgets per user or request.
Implement a simple Redis cache for repeated queries:
import hashlib
import redis
from openai import OpenAI
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
client = OpenAI()
def generate_summary_with_cache(text: str, user_id: str) -> str:
# Create a cache key from the input
cache_key = f"summary:{user_id}:{hashlib.md5(text.encode()).hexdigest()}"
# Check cache first
cached = redis_client.get(cache_key)
if cached:
return cached
# Call API if not cached
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": f"Summarize: {text}"
}
],
max_tokens=200
)
summary = response.choices[0].message.content
# Cache for 7 days
redis_client.setex(cache_key, 7 * 24 * 3600, summary)
return summaryEnforce token limits per request to avoid runaway costs:
def generate_with_limit(query: str, max_tokens: int = 500) -> str:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": query
}
],
max_tokens=min(max_tokens, 1000), # Hard cap
temperature=0.7
)
# Log token usage for monitoring
logger.info(
f"Tokens used - Input: {response.usage.prompt_tokens}, "
f"Output: {response.usage.completion_tokens}"
)
return response.choices[0].message.contentUse Anthropic's prompt caching feature if available for your model:
from anthropic import Anthropic
client = Anthropic()
def analyze_with_cache(documents: List[str], question: str) -> str:
# Large, stable context that may be reused
cached_context = "\n\n".join(documents)
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[
{
"type": "text",
"text": "You are a document analyst."
},
{
"type": "text",
"text": f"Documents: {cached_context}",
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{
"role": "user",
"content": question
}
]
)
return response.content[0].textMonitor usage and set alerts when costs exceed thresholds. Implement a rate limiter per user to prevent abuse:
from datetime import datetime, timedelta
def check_rate_limit(user_id: str, limit_per_hour: int = 100) -> bool:
key = f"rate:{user_id}:{datetime.utcnow().hour}"
count = redis_client.incr(key)
if count == 1:
redis_client.expire(key, 3600)
return count <= limit_per_hourCommon mistakes
- Hardcoding API keys — Always use environment variables or a secrets manager. Never commit keys to version control.
- Ignoring streaming for long operations — If your response takes more than a second or two, stream it. Users perceive this as faster.
- Not validating LLM output — Treat LLM responses as untrusted input. Parse and validate with Pydantic or a schema validator.
- No retry logic — API calls fail; use exponential backoff. Official SDKs often handle this, but verify your setup.
- Not monitoring costs — Log token usage for every request. Set up alerts before your bill surprises you.
- Mixing streaming and caching poorly — Cache full responses before streaming, not individual tokens. Mixing them adds complexity.
Key takeaways
- Use official OpenAI and Anthropic SDKs for type safety, error handling, and authentication out of the box.
- Stream responses to improve perceived latency and UX, especially for outputs longer than a few seconds.
- Validate LLM output with Pydantic models and structured output formats to prevent downstream errors.
- Cache responses in Redis, enforce token limits, and monitor usage to control costs in production.
- Log token usage and implement rate limiting per user to detect abuse and optimize spending.
- Handle errors gracefully; LLMs are probabilistic, so retries and fallback strategies are essential.
Comments
Loading comments…