Advanced24 min

Calling LLM APIs in Python

HTTP, JSON payloads, structured output, token accounting, and cost control — the real mechanics of a GenAI integration.

Why this matters in AI / ML / GenAI

Every GenAI feature is HTTP plus JSON underneath the SDK. Knowing the payload shape lets you switch providers, debug a 400, parse a tool call, and explain your bill. It is also the most common live-coding topic in GenAI interviews.

The request shape

Almost every chat provider — OpenAI, Anthropic, Bedrock, Gemini, vLLM, Ollama — accepts the same core structure:

  • model — which weights to run
  • messages — a list of {role, content} dicts with roles system, user, assistant
  • temperature — 0 for deterministic extraction, 0.7+ for creative writing
  • max_tokens — cap on the response length

The response carries the text plus a usage block with prompt_tokens and completion_tokens. That usage block is what your finance team eventually asks about, so log it from day one.

Set temperature=0 for anything you parse programmatically. Creative sampling in an extraction pipeline produces intermittent, unreproducible failures.

Structured output

Do not regex free-form prose. Ask for JSON and validate it.

Three layers of reliability, in order of strength:

  1. Ask for JSON in the prompt and json.loads the result.
  2. Use the provider's JSON mode or schema-constrained decoding.
  3. Validate against a pydantic model and retry once on failure with the validation error fed back in.

Always wrap parsing in try/except json.JSONDecodeError. Models occasionally wrap JSON in markdown fences — strip those before parsing.

Tokens, cost, and safety

Tokens are sub-word pieces. English averages roughly four characters per token, so len(text) / 4 is a workable estimate when tiktoken is not available.

Cost is (prompt_tokens * input_price + completion_tokens * output_price) / 1000. Long retrieved contexts are usually the expensive part, not the answer.

Four production habits:

  • Set an explicit timeout on every call — the default of "forever" will hang your service.
  • Retry 429 and 5xx with backoff; never retry a 400 or 401.
  • Cache identical prompts; repeated identical questions are common and free to serve from a cache.
  • Truncate context to a token budget before sending, or you will hit context-limit errors in production, not in testing.

Copy-paste examples

Copy into your own editor, or load one into the compiler below and press Run.

Build and inspect a request payload

Runs here — this is the exact JSON an OpenAI-compatible endpoint expects.

import json

def build_request(question, context, model="gpt-4.1-mini"):
    return {
        "model": model,
        "temperature": 0,
        "max_tokens": 300,
        "messages": [
            {"role": "system", "content": "Answer only from the context. If missing, say you do not know."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    }

payload = build_request("What is MLOps?", "MLOps is the practice of running ML in production.")
print(json.dumps(payload, indent=2))
print("estimated prompt tokens:", sum(len(m["content"]) for m in payload["messages"]) // 4)

Parse structured JSON output defensively

Models sometimes wrap JSON in markdown fences. Strip, parse, validate.

import json

def parse_model_json(raw, required_keys):
    text = raw.strip()
    if text.startswith("```"):
        lines = [ln for ln in text.splitlines() if not ln.startswith("```")]
        text = "\n".join(lines)
    try:
        data = json.loads(text)
    except json.JSONDecodeError as err:
        raise ValueError(f"model did not return valid JSON: {err}") from err
    missing = [k for k in required_keys if k not in data]
    if missing:
        raise ValueError(f"missing keys: {missing}")
    return data

good = '```json\n{"sentiment": "positive", "confidence": 0.92}\n```'
print(parse_model_json(good, ["sentiment", "confidence"]))

bad = "Sure! The sentiment is positive."
try:
    parse_model_json(bad, ["sentiment"])
except ValueError as err:
    print("rejected:", err)

Token estimate and cost calculator

Swap in tiktoken locally for exact counts; this estimate is close enough for budgeting.

PRICES_PER_1K = {
    "gpt-4.1-mini": {"input": 0.00015, "output": 0.0006},
    "gpt-4.1":      {"input": 0.002,   "output": 0.008},
}

def estimate_tokens(text):
    return max(1, len(text) // 4)

def estimate_cost(model, prompt, completion):
    price = PRICES_PER_1K[model]
    p_tokens = estimate_tokens(prompt)
    c_tokens = estimate_tokens(completion)
    cost = (p_tokens * price["input"] + c_tokens * price["output"]) / 1000
    return p_tokens, c_tokens, cost

context = "Retrieved context. " * 200
prompt = f"{context}\n\nQuestion: summarize the above."
answer = "The context repeats a placeholder sentence about retrieval."

for model in PRICES_PER_1K:
    p, c, cost = estimate_cost(model, prompt, answer)
    print(f"{model:14s} prompt={p:5d} completion={c:3d} cost=${cost:.6f}")

print("monthly at 100k calls (mini):", round(estimate_cost("gpt-4.1-mini", prompt, answer)[2] * 100_000, 2), "USD")

Production client with timeout, retry, and cache (run locally)

Copy into your project. pip install openai. Key comes from the environment.

# pip install openai
import functools
import json
import os
import time
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=30.0)

RETRYABLE = (429, 500, 502, 503, 504)

@functools.lru_cache(maxsize=512)
def ask(question: str, context: str = "", model: str = "gpt-4.1-mini") -> str:
    messages = [
        {"role": "system", "content": "Answer only from the context."},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
    ]
    for attempt in range(4):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                temperature=0,
                max_tokens=500,
            )
            usage = response.usage
            print(json.dumps({
                "event": "llm_call",
                "model": model,
                "prompt_tokens": usage.prompt_tokens,
                "completion_tokens": usage.completion_tokens,
            }))
            return response.choices[0].message.content
        except Exception as err:
            status = getattr(err, "status_code", None)
            if status not in RETRYABLE or attempt == 3:
                raise
            time.sleep(2 ** attempt)
    raise RuntimeError("unreachable")

Fit context into a token budget

Try it — in-browser Python

Lower max_context_tokens to 60 and watch chunks get dropped.

Output

Python runs in your browser. First run downloads the runtime.

Press Run (or Ctrl+Enter) to execute.

CPython in WebAssembly. Stdlib works. NumPy and pandas load on demand. No input(), no GPU, no network installs.

Takeaways

  • Every provider takes the same core payload: model, messages, temperature, max_tokens.
  • Use temperature=0 and validate JSON whenever you parse the output programmatically.
  • Always set timeouts, retry only transient errors, log usage, and budget context tokens.