Amazon Bedrock Prompt Caching and Batch Inference: Cut Coding-Agent Latency and Cost Without Stale Rules

TL;DR

• Put stable content first (tools, then system rules) and add a cachePoint after it. Dynamic content (user turns, retrieved chunks) goes after the checkpoint.
• Any change to the prefix is a cache miss, and changing tools invalidates the system and messages caches after it. Bedrock can’t serve an old prompt to a new request.
• The default TTL is 5 minutes and resets on each hit; some models support "ttl": "1h". Claude models allow up to 4 checkpoints and need a minimum prefix size.
• Read cacheReadInputTokens and cacheWriteInputTokens in the response. inputTokens counts only uncached tokens.
• Batch inference takes JSONL from S3 (InvokeModel or Converse format) and writes results back to S3. It doesn’t support prompt caching or tool calling.

How prompt caching works

• Implicit caching: Bedrock and the model automatically try to reuse eligible prefixes. No request changes are needed, but it’s best effort.
• Explicit caching: you mark where the reusable prefix ends with cache checkpoints (cachePoint in Converse, cache_control in Anthropic’s InvokeModel format).

• Exact prefix match. Prefixes must stay identical between requests; any change causes a miss.
• Order is tools, then system, then messages. The sections are chained, so changing tools invalidates everything after them.
• Minimum size. Each checkpoint needs a minimum number of tokens in the prefix before it (for Claude models in the current table, between 512 and 4,096 depending on the model). Below the minimum, the request still succeeds but nothing is cached.
• TTL. Many models use a 5-minute TTL that resets with every hit; some support a 1-hour TTL. Check the model card.
• No guarantee. Even with checkpoints, a hit isn’t guaranteed. Check the usage fields to see what happened.
• Billing. Cache reads are billed at the model’s cache-read rate. Cache writes can cost more than normal input, depending on the model. Output tokens are never discounted.

Step 1: Add cache checkpoints to agent calls

import json
import os

import boto3

brt = boto3.client("bedrock-runtime")
MODEL_ID = os.environ["AGENT_MODEL_ID"]  # a model or inference profile that supports caching

TOOLS = [
    {
        "toolSpec": {
            "name": "read_file",
            "description": "Read a file from the checked-out repository.",
            "inputSchema": {"json": {
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"],
            }},
        }
    },
    # ... more tool specs ...
    {"cachePoint": {"type": "default"}},  # checkpoint 1: after the tools
]


def build_system(rules_text: str) -> list[dict]:
    return [
        {"text": rules_text},                 # long, stable coding rules
        {"cachePoint": {"type": "default"}},  # checkpoint 2: after the rules
    ]


def agent_turn(rules_text: str, history: list[dict], user_text: str) -> dict:
    resp = brt.converse(
        modelId=MODEL_ID,
        system=build_system(rules_text),
        toolConfig={"tools": TOOLS},
        messages=history + [{"role": "user", "content": [{"text": user_text}]}],
        inferenceConfig={"maxTokens": 2048},
    )
    u = resp["usage"]
    read = u.get("cacheReadInputTokens", 0)
    write = u.get("cacheWriteInputTokens", 0)
    total_input = u["inputTokens"] + read + write  # inputTokens excludes cached tokens
    print(json.dumps({
        "metric": "bedrock_usage",
        "input_total": total_input,
        "cache_read": read,
        "cache_write": write,
        "cache_hit_ratio": round(read / total_input, 3) if total_input else 0,
        "output": u["outputTokens"],
    }))
    return resp

• Nothing that changes per request (timestamps, user names, request IDs, retrieved chunks) may appear before a checkpoint. One changing character in the rules section turns every call into a cache write.
• If your total prefix before a checkpoint is below the model’s minimum, it isn’t cached. Small system prompts may not benefit at all.
• For long sessions with pauses longer than 5 minutes, a model that supports "ttl": "1h" in the cachePoint keeps the cache alive. Don’t send the ttl field to models that only support 5 minutes; that can cause a ValidationException.
• Cross-Region inference works with caching, but at times of high demand it may lead to more cache writes.

Step 2: Measure whether caching pays off

• Without caching, a prefix of P tokens is billed P times the input rate on every call.
• With caching, the first call in a TTL window pays P times the cache-write rate, and each later hit pays P times the cache-read rate.

Step 3: Avoid stale coding rules

• The agent runtime loads rules from an old checkout or an old prompt version pin.
• An application-level cache (for example, a cache of answers or of assembled prompts) is keyed only by the question, not by the rules version.
• Old few-shot examples that use the removed API are still in the prompt.

import hashlib
import pathlib
import re


def rules_version(rules_dir: str) -> str:
    """Content hash of every rules file, so any edit produces a new version."""
    h = hashlib.sha256()
    for path in sorted(pathlib.Path(rules_dir).rglob("*.md")):
        h.update(str(path).encode())
        h.update(path.read_bytes())
    return h.hexdigest()[:16]


def app_cache_key(tenant: str, question: str, rules_ver: str, tools_ver: str) -> str:
    # Anything that changes behavior belongs in the key, and so does the tenant.
    return hashlib.sha256(f"{tenant}|{rules_ver}|{tools_ver}|{question}".encode()).hexdigest()


CANARIES = [
    {"prompt": "Add a function that loads a user record.",
     "must_not_match": r"LegacyUserRepo", "reason": "LegacyUserRepo removed in ADR-214"},
]


def run_canaries(ask) -> None:
    """Run after every rules change; fail the deploy if a removed pattern comes back."""
    for c in CANARIES:
        answer = ask(c["prompt"])
        if re.search(c["must_not_match"], answer):
            raise AssertionError(f"stale rule detected: {c['reason']}")

Step 4: Move offline work to batch inference

import json
import os

import boto3

s3 = boto3.client("s3")
bedrock = boto3.client("bedrock")
BUCKET = "agent-batch-jobs"
RULES = open("rules/system.md").read()


def submit_eval_batch(run_id: str, cases: list[dict]) -> str:
    lines = []
    for case in cases:
        lines.append(json.dumps({
            "recordId": case["id"],
            "modelInput": {
                "system": [{"text": RULES}],
                "messages": [{"role": "user", "content": [{"text": case["prompt"]}]}],
                "inferenceConfig": {"maxTokens": 1024},
            },
        }))
    key = f"evals/{run_id}/input.jsonl"
    s3.put_object(Bucket=BUCKET, Key=key, Body=("\n".join(lines) + "\n").encode())

    job = bedrock.create_model_invocation_job(
        jobName=f"eval-{run_id}",
        roleArn=os.environ["BATCH_ROLE_ARN"],
        modelId=os.environ["EVAL_MODEL_ID"],
        modelInvocationType="Converse",
        inputDataConfig={"s3InputDataConfig": {"s3Uri": f"s3://{BUCKET}/{key}"}},
        outputDataConfig={"s3OutputDataConfig": {"s3Uri": f"s3://{BUCKET}/evals/{run_id}/out/"}},
        timeoutDurationInHours=24,
    )
    return job["jobArn"]

import json

import boto3

s3 = boto3.client("s3")


def read_batch_results(bucket: str, prefix: str):
    """Yield (record_id, output, error) from every .out file except the manifest."""
    paginator = s3.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if key.endswith("manifest.json.out"):
                manifest = json.loads(s3.get_object(Bucket=bucket, Key=key)["Body"].read())
                print("records ok/err:", manifest["successRecordCount"], manifest["errorRecordCount"])
                continue
            if not key.endswith(".out"):
                continue
            body = s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode()
            for line in body.splitlines():
                rec = json.loads(line)
                yield rec.get("recordId"), rec.get("modelOutput"), rec.get("error")

Caching or batch: choose per workload

Worked example: one agent, two paths

1. Your review agent sends about 9,000 tokens of tool specs and rules on every turn (illustrative). You add checkpoints after the tools and after the rules.
2. The usage metric shows the first turn of each session writing to the cache and later turns reading from it. Sessions with long pauses show repeated writes, so you test the 1-hour TTL on a model that supports it.
3. A platform team renames a core package. The rules hash changes, the next calls show cache writes instead of reads (as expected), and the canary confirms the agent no longer suggests the old package.
4. The nightly eval of 5,000 cases moves from an on-demand loop to one batch job in the Converse format. The morning report joins outputs by recordId and lists the records that came back with errors.

Troubleshooting

Production checklist

• ☐ Checkpoints after tools and after system rules; nothing dynamic before them
• ☐ Prefix meets the model’s minimum tokens; TTL chosen per session pattern
• ☐ Cache read/write tokens and hit ratio logged per route
• ☐ Rules version (content hash) logged with each call and included in your own cache keys
• ☐ Canary prompts run after every rules or prompt-version change
• ☐ Untrusted, per-request text placed after the checkpoints
• ☐ Offline workloads on batch inference, joined by recordId, with EventBridge notifications
• ☐ Savings forecasts based on measured hit ratios and current pricing

FAQ

Can I use prompt caching inside a batch job?

Can Prompt Management prompts use caching?

Does caching change the model’s answers?

Why are my batch outputs out of order?

Official documentation

• Prompt caching for faster model inference (checkpoints, TTL, supported models)
• TokenUsage (cache usage fields)
• Process multiple prompts with batch inference
• Format and upload batch inference data
• Create a batch inference job
• Batch job statuses
• View the results of a batch inference job