The Messages Array — Core Concept
CRITICAL
Every modern LLM API is built around a conversation structure. You send an array of messages, each with a role and content.
Three Roles
systemInstructions for the model. Sets persona, constraints, behavior. User never sees this.
userInput from the human (or your code acting as the human).
assistantPrevious AI responses. Used to maintain multi-turn conversation context.
finish_reason Values
stopNormal completion ✓
lengthHit max_tokens — response cut off. Increase max_tokens. ✗
content_filterBlocked by safety system. Adjust prompt. ✗
tool_callsModel wants to call a function (agent mode). ✓
Key insight: The messages array IS the conversation history. You send the full history on every request — the API is stateless. The model has no memory except what you send in the array.
OpenAI Chat Completions API
from openai import OpenAI
client = OpenAI() # reads OPENAI_API_KEY from env
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a helpful coding assistant."},
{"role": "user", "content": "Explain what a closure is in Python."}
],
temperature=0.7,
max_tokens=500,
)
print(response.choices[0].message.content)
print(f"finish_reason: {response.choices[0].finish_reason}")
print(f"Tokens: {response.usage.prompt_tokens} in / {response.usage.completion_tokens} out")
Key Response Fields
response.choices[0].message.content # The text output
response.choices[0].finish_reason # "stop", "length", "content_filter"
response.usage.prompt_tokens # Input tokens consumed
response.usage.completion_tokens # Output tokens generated
response.usage.total_tokens # Sum (what you pay for)
response.model # Exact model version used
Anthropic Claude API — Key Differences
import anthropic
client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY from env
message = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=500,
system="You are a helpful coding assistant.", # ← system is a TOP-LEVEL param
messages=[
{"role": "user", "content": "Explain what a closure is in Python."}
],
)
print(message.content[0].text)
print(f"Tokens: in={message.usage.input_tokens}, out={message.usage.output_tokens}")
Key difference from OpenAI: The system prompt is a separate top-level parameter, NOT part of the messages array. The message structure is otherwise very similar.
import google.generativeai as genai
import os
genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
model = genai.GenerativeModel(
model_name="gemini-1.5-flash",
system_instruction="You are a helpful coding assistant."
)
response = model.generate_content("Explain what a closure is in Python.")
print(response.text)
# Multi-turn chat session
chat = model.start_chat()
r1 = chat.send_message("What is Python?")
r2 = chat.send_message("What about closures?") # remembers context
Cost Reference (per 1M tokens)
Memorize This
| Model | Input | Output | Best For |
| gpt-4o-mini | $0.15 | $0.60 | Default choice — most tasks, high volume |
| gpt-4o | $5.00 | $15.00 | Complex reasoning, quality critical |
| claude-3-5-sonnet | $3.00 | $15.00 | Long context, nuanced reasoning |
| claude-3-haiku | $0.25 | $1.25 | Fast, cheap, simple tasks |
| gemini-1.5-flash | $0.075 | $0.30 | Cheapest option, free tier available |
$10 of gpt-4o-mini = ~39,000 API calls at 500 input + 300 output tokens. This is why it's the default.
Streaming Responses
Always Do This in Production
Without streaming, users wait for the full response. With streaming, text appears token by token — dramatically better UX.
# OpenAI streaming
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Write a poem about Python."}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
print() # newline when done
# Anthropic streaming
with client.messages.stream(
model="claude-3-haiku-20240307",
max_tokens=500,
messages=[{"role": "user", "content": "Write a poem about Python."}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
Rule: Always stream in user-facing applications. Only skip streaming for background batch jobs or when you need the full response before acting on it.
Error Handling — What to Retry vs What to Fix
import time
from openai import OpenAI, RateLimitError, APITimeoutError, APIError
client = OpenAI()
def call_with_retry(messages, model="gpt-4o-mini", max_retries=3):
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model=model, messages=messages, max_tokens=500, timeout=30
)
return response.choices[0].message.content
except RateLimitError:
wait = 2 ** attempt # exponential backoff: 1s, 2s, 4s
print(f"Rate limited. Waiting {wait}s...")
time.sleep(wait)
except APITimeoutError:
if attempt == max_retries - 1: raise
time.sleep(1)
except APIError as e:
if e.status_code >= 500: # server error → retry
time.sleep(1)
else:
raise # client error → fix your code, don't retry
raise Exception("Max retries exceeded")
| Error | Cause | Action |
| RateLimitError (429) | Too many requests | Exponential backoff, retry ✓ |
| APITimeoutError | Request took too long | Retry with backoff ✓ |
| AuthenticationError | Bad API key | Fix your key, NEVER retry ✗ |
| BadRequestError (400) | Invalid params or prompt | Fix your code, NEVER retry ✗ |
| APIError (500+) | Provider server error | Retry once or twice ✓ |
Multi-Provider Wrapper — Production Pattern
In production you want to swap providers without changing application code. Abstract behind a common interface:
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class LLMResponse:
content: str
input_tokens: int
output_tokens: int
model: str
class LLMProvider(ABC):
@abstractmethod
def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
pass
class OpenAIProvider(LLMProvider):
def __init__(self, model: str = "gpt-4o-mini"):
from openai import OpenAI
self.client = OpenAI()
self.model = model
def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
**kwargs
)
return LLMResponse(
content=resp.choices[0].message.content,
input_tokens=resp.usage.prompt_tokens,
output_tokens=resp.usage.completion_tokens,
model=resp.model
)
# Usage — swap provider without changing any other code:
provider = OpenAIProvider()
# provider = AnthropicProvider() # just change this line
result = provider.complete(system="You are helpful.", user="Hello!")
print(result.content)
Fallback chain: Try OpenAI → if fails, try Anthropic → if fails, try Gemini. Use try/except around each provider call. Log which provider was used in production.
Token Counting with tiktoken
import tiktoken
def count_tokens(messages, model="gpt-4o-mini") -> int:
enc = tiktoken.encoding_for_model(model)
total = 3 # every reply is primed with <|start|>assistant<|message|>
for msg in messages:
total += 4 # every message has role/content overhead
total += len(enc.encode(msg["content"]))
return total
def estimate_cost(messages, model="gpt-4o-mini", expected_output=300) -> float:
RATES = {"gpt-4o-mini": (0.15, 0.60), "gpt-4o": (5.00, 15.00)}
in_rate, out_rate = RATES.get(model, (0.15, 0.60))
in_tokens = count_tokens(messages, model)
cost = (in_tokens * in_rate + expected_output * out_rate) / 1_000_000
return round(cost, 6)
# Check before expensive batch jobs:
print(f"Estimated cost: ${estimate_cost(messages):.4f}")
Rule of thumb: 1 token ≈ 0.75 words in English. 1,000 tokens ≈ 750 words ≈ 1.5 pages of text. Always count before running batch jobs on large datasets.
API Key Security — Non-Negotiable Rules
NEVER hardcode API keys in source code. They leak into git history, error messages, and logs. One exposed key costs hundreds of dollars in minutes.
# .env file (add to .gitignore immediately)
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AI...
# Python — correct way
from dotenv import load_dotenv
import os
load_dotenv() # reads .env into environment
api_key = os.getenv("OPENAI_API_KEY") # reads from environment
client = OpenAI(api_key=api_key)
# Or even simpler — OpenAI reads from env automatically:
client = OpenAI() # reads OPENAI_API_KEY from environment
# .gitignore — add this line
.env
Cheatsheet — Quick Reference
OpenAI Minimal
from openai import OpenAI
client = OpenAI()
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "..."},
{"role": "user", "content": "..."},
],
temperature=0.7,
max_tokens=500,
)
print(resp.choices[0].message.content)
Anthropic Minimal
import anthropic
client = anthropic.Anthropic()
msg = client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=500,
system="...",
messages=[
{"role": "user", "content": "..."},
],
)
print(msg.content[0].text)
Gemini Minimal
import google.generativeai as genai
genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
model = genai.GenerativeModel("gemini-1.5-flash")
print(model.generate_content("...").text)
Streaming (OpenAI)
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content,
end="", flush=True)