Cost, latency, and limits

PRICES_PER_MILLION = {          # (input $, output $) - example numbers only
    "small-fast-model": (0.15, 0.60),
    "mid-model": (0.40, 1.60),
    "frontier-model": (2.50, 10.00),
}


def call_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    price_in, price_out = PRICES_PER_MILLION[model]
    return (input_tokens * price_in + output_tokens * price_out) / 1_000_000


# a RAG answer: 3,000 tokens of context + question, 400-token answer
for model in PRICES_PER_MILLION:
    one = call_cost(model, 3_000, 400)
    monthly = one * 20_000 * 30            # 20k requests per day
    print(f"{model:<17} ${one:.5f} per call   ${monthly:>9,.2f} per month")

total latency ≈ network + queue + TTFT (prefill of input) + output_tokens / tokens_per_second

import time

from openai import OpenAI

client = OpenAI()

start = time.perf_counter()
first_token_at = None
pieces = []

stream = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "Explain RAG in 80 words."}],
    stream=True,
    stream_options={"include_usage": True},     # final chunk carries token usage
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage
    if chunk.choices and chunk.choices[0].delta.content:
        if first_token_at is None:
            first_token_at = time.perf_counter()
        pieces.append(chunk.choices[0].delta.content)
end = time.perf_counter()

ttft = first_token_at - start
gen_time = end - first_token_at
print(f"TTFT: {ttft * 1000:.0f} ms")
print(f"total: {(end - start) * 1000:.0f} ms")
print(f"speed: {usage.completion_tokens / gen_time:.0f} tokens/s")

x-ratelimit-limit-requests: 5000
x-ratelimit-remaining-requests: 4987
x-ratelimit-limit-tokens: 2000000
x-ratelimit-remaining-tokens: 1964210
x-ratelimit-reset-tokens: 1.07s

import time
from concurrent.futures import ThreadPoolExecutor


def fake_call(prompt: str) -> str:
    time.sleep(0.2)                      # stand-in for network + generation time
    return prompt.upper()


prompts = [f"summarise doc {i}" for i in range(10)]

start = time.perf_counter()
sequential = [fake_call(p) for p in prompts]
print(f"sequential: {time.perf_counter() - start:.1f}s")

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:     # at most 5 requests in flight
    parallel = list(pool.map(fake_call, prompts))
print(f"parallel (5 workers): {time.perf_counter() - start:.1f}s")
print(parallel[:2], parallel == sequential)

import json
import time
from dataclasses import asdict, dataclass

PRICES_PER_MILLION = {"small-fast-model": (0.15, 0.60)}


@dataclass
class LLMCallLog:
    feature: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    finish_reason: str
    cost_usd: float


def log_call(feature, model, usage, latency_s, finish_reason) -> LLMCallLog:
    price_in, price_out = PRICES_PER_MILLION[model]
    cost = (usage["prompt_tokens"] * price_in + usage["completion_tokens"] * price_out) / 1e6
    entry = LLMCallLog(feature, model, usage["prompt_tokens"], usage["completion_tokens"],
                       round(latency_s * 1000, 1), finish_reason, round(cost, 6))
    print(json.dumps(asdict(entry)))
    return entry


t0 = time.perf_counter()
time.sleep(0.05)                                   # pretend API call
fake_usage = {"prompt_tokens": 1200, "completion_tokens": 180}
entry = log_call("support-bot", "small-fast-model", fake_usage, time.perf_counter() - t0, "stop")
print(entry.cost_usd)

import hashlib
import json

CACHE: dict[str, str] = {}
calls = {"api": 0}


def fake_llm(messages, model, temperature):
    calls["api"] += 1
    return f"answer to: {messages[-1]['content']}"


def cached_chat(messages, model="small-fast-model", temperature=0.0) -> str:
    key_src = json.dumps({"m": messages, "model": model, "t": temperature}, sort_keys=True)
    key = hashlib.sha256(key_src.encode()).hexdigest()
    if key not in CACHE:
        CACHE[key] = fake_llm(messages, model, temperature)
    return CACHE[key]


q = [{"role": "user", "content": "What are your opening hours?"}]
for _ in range(3):
    print(cached_chat(q))
print("API calls made:", calls["api"])