OpenAI API Production Patterns

Structured Outputs with Pydantic

from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI()

class ProductReview(BaseModel):
    sentiment: str = Field(description="positive, negative, or neutral")
    score: int = Field(description="Rating 1-5", ge=1, le=5)
    key_themes: list[str]
    summary: str
    requires_response: bool

def analyze_review(text: str) -> ProductReview:
    resp = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Analyze customer reviews."},
            {"role": "user", "content": f"Analyze: {text}"},
        ],
        response_format=ProductReview,
        temperature=0.1,
    )
    return resp.choices[0].message.parsed

Function Calling

import json

tools = [{
    "type": "function",
    "function": {
        "name": "search_products",
        "description": "Search product catalog",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "category": {"type": "string", "enum": ["electronics", "clothing"]},
                "max_price": {"type": "number"},
            },
            "required": ["query"],
        },
    },
}]

def agent_loop(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    while True:
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages, tools=tools
        )
        choice = resp.choices[0]
        messages.append(choice.message)
        if choice.finish_reason == "stop":
            return choice.message.content
        for tc in choice.message.tool_calls:
            result = execute_tool(tc.function.name, json.loads(tc.function.arguments))
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

Streaming

async def stream_chat(messages):
    stream = await async_client.chat.completions.create(
        model="gpt-4o", messages=messages, stream=True
    )
    full_response = ""
    async for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            full_response += delta.content
            yield delta.content
    return full_response

Prompt Caching

# GPT-4o automatically caches prompts >= 1024 tokens
# For long system prompts, put stable content first

system_prompt = """[Long stable system context - 2000 tokens]
...product documentation...
...rules and guidelines..."""

def cached_query(user_question: str) -> str:
    # System prompt gets cached after first call
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},  # cached
            {"role": "user", "content": user_question},    # dynamic
        ],
    ).choices[0].message.content

Batch API for Cost Savings

import json

requests = [
    {"custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions",
     "body": {"model": "gpt-4o-mini", "messages": [
         {"role": "user", "content": f"Summarize: {text}"}
     ]}}
    for i, text in enumerate(large_text_list)
]

# Write batch file
with open("batch_requests.jsonl", "w") as f:
    for req in requests:
        f.write(json.dumps(req) + "
")

# Submit batch (50% cost reduction, 24h window)
batch_input_file = client.files.create(file=open("batch_requests.jsonl", "rb"), purpose="batch")
batch = client.batches.create(
    input_file_id=batch_input_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
print(f"Batch ID: {batch.id}")

Cost Optimization Tips