Choosing a Foundation Model on Amazon Bedrock: A Practical Comparison

import csv
import boto3
from botocore.exceptions import ClientError

REGION = "us-east-1"  # use your region
client = boto3.client("bedrock-runtime", region_name=REGION)

# Copy exact IDs from each model's card, inference profile prefix included.
MODELS = {
    "Model A": "MODEL_ID_A",
    "Model B": "MODEL_ID_B",
    "Model C": "MODEL_ID_C",
}

# USD per 1 million tokens: (input, output). Copy from the Bedrock pricing page.
PRICES = {
    "Model A": (0.0, 0.0),
    "Model B": (0.0, 0.0),
    "Model C": (0.0, 0.0),
}

PROMPTS = {
    "summarise": "Summarise this in three bullet points:\n\n<your text>",
    "extract": "Return JSON with name, date and amount from:\n\n<your text>",
    "reason": "<your reasoning problem>",
    "rewrite": "Rewrite this in a friendly tone:\n\n<your text>",
    "explain": "Explain what this does in plain English:\n\n<your code>",
}

RUNS = 3

def ask(model_id, prompt):
    return client.converse(
        modelId=model_id,
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        inferenceConfig={"maxTokens": 500, "temperature": 0},
    )

rows = []
for model_name, model_id in MODELS.items():
    price_in, price_out = PRICES[model_name]
    for prompt_name, prompt in PROMPTS.items():
        for run in range(1, RUNS + 1):
            try:
                r = ask(model_id, prompt)
            except ClientError as e:
                rows.append([model_name, prompt_name, run, e.response["Error"]["Code"], "", "", "", ""])
                continue
            text = r["output"]["message"]["content"][0]["text"]
            tokens_in = r["usage"]["inputTokens"]
            tokens_out = r["usage"]["outputTokens"]
            latency = r["metrics"]["latencyMs"]
            cost = tokens_in / 1e6 * price_in + tokens_out / 1e6 * price_out
            rows.append([model_name, prompt_name, run, "ok", latency, tokens_in, tokens_out, round(cost, 6)])
            # Save the answer so you can score it later.
            with open(f"answers_{model_name}_{prompt_name}_{run}.txt", "w") as f:
                f.write(text)

with open("results.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["model", "prompt", "run", "status", "latency_ms", "in_tokens", "out_tokens", "cost_usd"])
    w.writerows(rows)

print("Done. See results.csv and the answers_*.txt files.")

import csv
import statistics
from collections import defaultdict

latency = defaultdict(list)
cost = defaultdict(list)
for row in csv.DictReader(open("results.csv")):
    if row["status"] != "ok":
        continue
    latency[row["model"]].append(float(row["latency_ms"]))
    cost[row["model"]].append(float(row["cost_usd"]))

quality = defaultdict(list)
for row in csv.DictReader(open("scores.csv")):
    quality[row["model"]].append(float(row["quality"]))

print("| Model | Average quality (1 to 5) | Median latency (ms) | Cost per 100 calls (USD) |")
print("|---|---|---|---|")
for model in latency:
    q = f"{statistics.mean(quality[model]):.1f}" if quality[model] else ""
    print(f"| {model} | {q} | {statistics.median(latency[model]):.0f} | {statistics.mean(cost[model]) * 100:.2f} |")

from concurrent.futures import ThreadPoolExecutor

def burst(model_id, n=20, workers=10):
    def one(_):
        try:
            ask(model_id, "Say hello in five words.")
            return "ok"
        except ClientError as e:
            return e.response["Error"]["Code"]
    with ThreadPoolExecutor(max_workers=workers) as pool:
        results = list(pool.map(one, range(n)))
    return {code: results.count(code) for code in set(results)}

print("Regional:", burst("REGIONAL_MODEL_ID"))
print("Profile: ", burst("PROFILE_MODEL_ID"))