How to Cut AI Agent Cost: Prompt Caching and Model Routing

route=claude-sonnet-5-5 cost=$0.01305fields=PASSpriority=high needs_human=Truepromises_action=Falseroute=claude-haiku-5-5 cost=$0.00038fields=PASSpriority=medium needs_human=Falsepromises_action=Truechecked=2 failed=0

pip install anthropic==0.120.0

python -c "import anthropic; print(anthropic.__version__)"

0.120.0

import jsonimport anthropicSONNET = "claude-sonnet-5-5"MAX_TURNS = 6SYSTEM = (    "You triage support tickets for a SaaS billing product. Use the tools to look up "    "the customer and the billing policy before deciding. "    "Then reply with one JSON object "    "only: {\"category\": str, \"priority\": \"low|medium|high\", \"reply_draft\": str, "    "\"needs_human\": bool}."    "\n\nSet needs_human to true only when one of these holds: the customer asks "    "for money back or an exception the policy docs do not cover; the docs have no "    "answer to the question; or the account needs a manual change by staff. "    "Otherwise set it to false, even when the topic sounds sensitive. When the docs "    "state the rule that applies, answer from the rule.")TOOLS = [    {        "name": "lookup_customer",        "description": (            "Look up a customer's plan, account status, and recent invoices "            "by customer id."        ),        "input_schema": {            "type": "object",            "properties": {"customer_id": {"type": "string"}},            "required": ["customer_id"],            "additionalProperties": False,        },    },    {        "name": "search_docs",        "description": "Search the billing policy docs. Returns matching snippets.",        "input_schema": {            "type": "object",            "properties": {"query": {"type": "string"}},            "required": ["query"],            "additionalProperties": False,        },    },]CUSTOMERS = {    "c1": {        "plan": "Pro", "status": "active",        "invoices": [{"id": "INV-101", "amount": 49, "status": "paid"}],    },    "c2": {        "plan": "Team", "status": "past_due",        "invoices": [{"id": "INV-204", "amount": 199, "status": "open"}],    },}DOCS = {    "refund": (        "Refunds: full refund within 14 days of a charge; after 14 days, "        "prorated credit only, approved by a human."    ),    "past due": (        "Past due: accounts are suspended 7 days after an invoice stays open; "        "a retry runs on day 3 and day 5."    ),    "cancel": (        "Cancellation: takes effect at period end; data is kept 30 days; "        "reactivation within 30 days is free."    ),    "upgrade": "Upgrades are prorated immediately; downgrades apply at the next renewal.",}def dispatch(name, args):    if name == "lookup_customer":        return json.dumps(CUSTOMERS.get(args.get("customer_id"), {"error": "not found"}))    if name == "search_docs":        q = args.get("query", "").lower()        hits = [v for k, v in DOCS.items() if any(w in q for w in k.split())]        return json.dumps(hits or ["no matching policy"])    return json.dumps({"error": f"unknown tool {name}"})def triage(client, model, ticket):    messages = [{        "role": "user",        "content": f"Customer id: {ticket['customer_id']}\n\nTicket: {ticket['text']}",    }]    for _ in range(MAX_TURNS):        r = client.messages.create(            model=model, max_tokens=1024, system=SYSTEM, tools=TOOLS, messages=messages,        )        messages.append({"role": "assistant", "content": r.content})        if r.stop_reason != "tool_use":            text = "".join(b.text for b in r.content if b.type == "text")            return json.loads(text)        results = [            {                "type": "tool_result",                "tool_use_id": b.id,                "content": dispatch(b.name, b.input),            }            for b in r.content if b.type == "tool_use"        ]        messages.append({"role": "user", "content": results})    raise RuntimeError(f"no verdict after {MAX_TURNS} turns")if __name__ == "__main__":    client = anthropic.Anthropic()    ticket = {        "customer_id": "c2",        "text": "Our card failed and now the dashboard is locked. What do we do?",    }    print(json.dumps(triage(client, SONNET, ticket), indent=2))

python triage.py

def triage(client, model, ticket, cache=True):    messages = [{        "role": "user",        "content": f"Customer id: {ticket['customer_id']}\n\nTicket: {ticket['text']}",    }]    usage = {"calls": 0, "input": 0, "output": 0, "cache_write": 0, "cache_read": 0}    extra = {"cache_control": {"type": "ephemeral"}} if cache else {}    for _ in range(MAX_TURNS):        r = client.messages.create(            model=model, max_tokens=1024, system=SYSTEM, tools=TOOLS,            messages=messages, **extra,        )        usage["calls"] += 1        usage["input"] += r.usage.input_tokens        usage["output"] += r.usage.output_tokens        usage["cache_write"] += r.usage.cache_creation_input_tokens or 0        usage["cache_read"] += r.usage.cache_read_input_tokens or 0        messages.append({"role": "assistant", "content": r.content})        if r.stop_reason != "tool_use":            text = "".join(b.text for b in r.content if b.type == "text")            return json.loads(text), usage        results = [            {                "type": "tool_result",                "tool_use_id": b.id,                "content": dispatch(b.name, b.input),            }            for b in r.content if b.type == "tool_use"        ]        messages.append({"role": "user", "content": results})    raise RuntimeError(f"no verdict after {MAX_TURNS} turns")

    verdict, usage = triage(client, SONNET, ticket)    print(json.dumps(verdict, indent=2))    print(usage)

OUTPUT_FORMAT = {    "type": "json_schema",    "schema": {        "type": "object",        "properties": {            "category": {"type": "string"},            "priority": {"type": "string", "enum": ["low", "medium", "high"]},            "reply_draft": {"type": "string"},            "needs_human": {"type": "boolean"},        },        "required": ["category", "priority", "reply_draft", "needs_human"],        "additionalProperties": False,    },}

        r = client.messages.create(            model=model, max_tokens=1024, system=SYSTEM, tools=TOOLS,            messages=messages, output_config={"format": OUTPUT_FORMAT}, **extra,        )

HAIKU = "claude-haiku-5-5"PRICES = {  # dollars per million tokens: input, output, cache write, cache read    SONNET: (2.00, 10.00, 2.50, 0.10),  # cache read is 0.05x base for Sonnet 5.5    HAIKU: (0.10, 0.50, 0.125, 0.01),}def cost(model, usage):    pi, po, pw, pr = PRICES[model]    return (usage["input"] * pi + usage["output"] * po            + usage["cache_write"] * pw + usage["cache_read"] * pr) / 1e6def triage_with_cascade(client, ticket):    verdict, usage_h = triage(client, HAIKU, ticket)    total = cost(HAIKU, usage_h)    route = HAIKU    if verdict["priority"] == "high":        verdict, usage_s = triage(client, SONNET, ticket)        total += cost(SONNET, usage_s)        route = SONNET    return verdict, route, total

PRIORITIES = {"low", "medium", "high"}PROMISE = re.compile(    r"process(ing)? (it|the refund)|will process|we'?ll (process|issue)", re.I)def check_verdict(verdict):    """Return the names of the fields that fail a code check. An empty list means pass."""    problems = []    if not isinstance(verdict.get("category"), str) or not verdict["category"].strip():        problems.append("category")    if verdict.get("priority") not in PRIORITIES:        problems.append("priority")    if not isinstance(verdict.get("needs_human"), bool):        problems.append("needs_human")    reply = verdict.get("reply_draft")    if not isinstance(reply, str) or not reply.strip():        problems.append("reply_draft")    return problemsdef promises_action(verdict):    """True when the reply promises an action the script cannot perform,    such as "we will process" a refund. Past tense is not caught."""    return bool(PROMISE.search(verdict.get("reply_draft", "")))

if __name__ == "__main__":    client = anthropic.Anthropic()    tickets = [        {            "customer_id": "c2",            "text": "Our card failed and now the dashboard is locked. What do we do?",        },        {            "customer_id": "c1",            "text": "Can I get a refund for my charge from yesterday?",        },    ]    failures = 0    for ticket in tickets:        verdict, route, total = triage_with_cascade(client, ticket)        problems = check_verdict(verdict)        failures += bool(problems)        print(f"route={route} cost=${total:.5f}")        print(f"fields={'PASS' if not problems else 'FAIL: ' + ', '.join(problems)}")        print(f"priority={verdict['priority']} needs_human={verdict['needs_human']}")        print(f"promises_action={promises_action(verdict)}")    print(f"checked={len(tickets)} failed={failures}")    raise SystemExit(1 if failures else 0)

route=claude-sonnet-5-5 cost=$0.01305fields=PASSpriority=high needs_human=Truepromises_action=Falseroute=claude-haiku-5-5 cost=$0.00038fields=PASSpriority=medium needs_human=Falsepromises_action=Truechecked=2 failed=0

import jsonimport reimport anthropicSONNET = "claude-sonnet-5-5"HAIKU = "claude-haiku-5-5"MAX_TURNS = 6SYSTEM = (    "You triage support tickets for a SaaS billing product. Use the tools to look up "    "the customer and the billing policy before deciding. "    "Then reply with one JSON object "    "only: {\"category\": str, \"priority\": \"low|medium|high\", \"reply_draft\": str, "    "\"needs_human\": bool}."    "\n\nSet needs_human to true only when one of these holds: the customer asks "    "for money back or an exception the policy docs do not cover; the docs have no "    "answer to the question; or the account needs a manual change by staff. "    "Otherwise set it to false, even when the topic sounds sensitive. When the docs "    "state the rule that applies, answer from the rule.")TOOLS = [    {        "name": "lookup_customer",        "description": (            "Look up a customer's plan, account status, and recent invoices "            "by customer id."        ),        "input_schema": {            "type": "object",            "properties": {"customer_id": {"type": "string"}},            "required": ["customer_id"],            "additionalProperties": False,        },    },    {        "name": "search_docs",        "description": "Search the billing policy docs. Returns matching snippets.",        "input_schema": {            "type": "object",            "properties": {"query": {"type": "string"}},            "required": ["query"],            "additionalProperties": False,        },    },]CUSTOMERS = {    "c1": {        "plan": "Pro", "status": "active",        "invoices": [{"id": "INV-101", "amount": 49, "status": "paid"}],    },    "c2": {        "plan": "Team", "status": "past_due",        "invoices": [{"id": "INV-204", "amount": 199, "status": "open"}],    },}DOCS = {    "refund": (        "Refunds: full refund within 14 days of a charge; after 14 days, "        "prorated credit only, approved by a human."    ),    "past due": (        "Past due: accounts are suspended 7 days after an invoice stays open; "        "a retry runs on day 3 and day 5."    ),    "cancel": (        "Cancellation: takes effect at period end; data is kept 30 days; "        "reactivation within 30 days is free."    ),    "upgrade": "Upgrades are prorated immediately; downgrades apply at the next renewal.",}def dispatch(name, args):    if name == "lookup_customer":        return json.dumps(CUSTOMERS.get(args.get("customer_id"), {"error": "not found"}))    if name == "search_docs":        q = args.get("query", "").lower()        hits = [v for k, v in DOCS.items() if any(w in q for w in k.split())]        return json.dumps(hits or ["no matching policy"])    return json.dumps({"error": f"unknown tool {name}"})OUTPUT_FORMAT = {    "type": "json_schema",    "schema": {        "type": "object",        "properties": {            "category": {"type": "string"},            "priority": {"type": "string", "enum": ["low", "medium", "high"]},            "reply_draft": {"type": "string"},            "needs_human": {"type": "boolean"},        },        "required": ["category", "priority", "reply_draft", "needs_human"],        "additionalProperties": False,    },}PRICES = {  # dollars per million tokens: input, output, cache write, cache read    SONNET: (2.00, 10.00, 2.50, 0.10),  # cache read is 0.05x base for Sonnet 5.5    HAIKU: (0.10, 0.50, 0.125, 0.01),}def triage(client, model, ticket, cache=True):    messages = [{        "role": "user",        "content": f"Customer id: {ticket['customer_id']}\n\nTicket: {ticket['text']}",    }]    usage = {"calls": 0, "input": 0, "output": 0, "cache_write": 0, "cache_read": 0}    extra = {"cache_control": {"type": "ephemeral"}} if cache else {}    for _ in range(MAX_TURNS):        r = client.messages.create(            model=model, max_tokens=1024, system=SYSTEM, tools=TOOLS,            messages=messages, output_config={"format": OUTPUT_FORMAT}, **extra,        )        usage["calls"] += 1        usage["input"] += r.usage.input_tokens        usage["output"] += r.usage.output_tokens        usage["cache_write"] += r.usage.cache_creation_input_tokens or 0        usage["cache_read"] += r.usage.cache_read_input_tokens or 0        messages.append({"role": "assistant", "content": r.content})        if r.stop_reason != "tool_use":            text = "".join(b.text for b in r.content if b.type == "text")            return json.loads(text), usage        results = [            {                "type": "tool_result",                "tool_use_id": b.id,                "content": dispatch(b.name, b.input),            }            for b in r.content if b.type == "tool_use"        ]        messages.append({"role": "user", "content": results})    raise RuntimeError(f"no verdict after {MAX_TURNS} turns")def cost(model, usage):    pi, po, pw, pr = PRICES[model]    return (usage["input"] * pi + usage["output"] * po            + usage["cache_write"] * pw + usage["cache_read"] * pr) / 1e6def triage_with_cascade(client, ticket):    verdict, usage_h = triage(client, HAIKU, ticket)    total = cost(HAIKU, usage_h)    route = HAIKU    if verdict["priority"] == "high":        verdict, usage_s = triage(client, SONNET, ticket)        total += cost(SONNET, usage_s)        route = SONNET    return verdict, route, totalPRIORITIES = {"low", "medium", "high"}PROMISE = re.compile(    r"process(ing)? (it|the refund)|will process|we'?ll (process|issue)", re.I)def check_verdict(verdict):    """Return the names of the fields that fail a code check. An empty list means pass."""    problems = []    if not isinstance(verdict.get("category"), str) or not verdict["category"].strip():        problems.append("category")    if verdict.get("priority") not in PRIORITIES:        problems.append("priority")    if not isinstance(verdict.get("needs_human"), bool):        problems.append("needs_human")    reply = verdict.get("reply_draft")    if not isinstance(reply, str) or not reply.strip():        problems.append("reply_draft")    return problemsdef promises_action(verdict):    """True when the reply promises an action the script cannot perform,    such as "we will process" a refund. Past tense is not caught."""    return bool(PROMISE.search(verdict.get("reply_draft", "")))if __name__ == "__main__":    client = anthropic.Anthropic()    tickets = [        {            "customer_id": "c2",            "text": "Our card failed and now the dashboard is locked. What do we do?",        },        {            "customer_id": "c1",            "text": "Can I get a refund for my charge from yesterday?",        },    ]    failures = 0    for ticket in tickets:        verdict, route, total = triage_with_cascade(client, ticket)        problems = check_verdict(verdict)        failures += bool(problems)        print(f"route={route} cost=${total:.5f}")        print(f"fields={'PASS' if not problems else 'FAIL: ' + ', '.join(problems)}")        print(f"priority={verdict['priority']} needs_human={verdict['needs_human']}")        print(f"promises_action={promises_action(verdict)}")    print(f"checked={len(tickets)} failed={failures}")    raise SystemExit(1 if failures else 0)

CUSTOMERS_EXTRA = {    "c3": {        "plan": "Free",        "status": "active",        "invoices": []    },    "c4": {        "plan": "Enterprise",        "status": "active",        "invoices": [            {                "id": "INV-330",                "amount": 2400,                "status": "paid"            }        ]    },    "c5": {        "plan": "Pro",        "status": "cancelled",        "invoices": [            {                "id": "INV-150",                "amount": 49,                "status": "refunded"            }        ]    },    "c6": {        "plan": "Pro",        "status": "active",        "invoices": [            {                "id": "INV-600",                "amount": 49,                "status": "paid",                "age_days": 90            },            {                "id": "INV-601",                "amount": 49,                "status": "paid",                "age_days": 1            }        ]    },    "c7": {        "plan": "Enterprise",        "status": "active",        "invoices": [            {                "id": "INV-500",                "amount": 12000,                "status": "open",                "overdue_days": 40            }        ]    },    "c8": {        "plan": "Free",        "status": "active",        "invoices": []    },    "c9": {        "plan": "Team",        "status": "past_due",        "invoices": [            {                "id": "INV-900",                "amount": 199,                "status": "open"            }        ]    },    "c10": {        "plan": "Pro",        "status": "active",        "invoices": [            {                "id": "INV-1000",                "amount": 49,                "status": "paid"            }        ]    }}

# measure.py: run the 20 labeled tickets through the cascade and score them.import anthropicimport triagefrom triage import CUSTOMERS_EXTRA, triage_with_cascadetriage.CUSTOMERS.update(CUSTOMERS_EXTRA)TICKETS = [    ("t1", "c1",        "Hi, I was charged twice this month. Can I get one back?"),    ("t2", "c2",        "Our card failed and now the dashboard is locked. What do we "        "do?"),    ("t3", "c3",        "How do I export my data to CSV?"),    ("t4", "c4",        "We want to add 20 seats mid-cycle. What does that cost?"),    ("t5", "c5",        "I cancelled last month but want my account back. Is that "        "possible?"),    ("t6", "c6",        "I want a refund for all the charges from last quarter."),    ("t7", "c7",        "Invoice 500 is unpaid. If it isn't resolved today we are "        "disputing it with our bank."),    ("t8", "c8",        "If I upgrade to Pro today, do I pay the full month or just "        "the rest of it?"),    ("t9", "c9",        "We already paid this, see the attached bank receipt. Please "        "stop the lockout."),    ("t10", "c10",        "Under GDPR I want all my personal data deleted. Please "        "confirm when it's done."),    ("t11", "c2",        "Can you move us from Team to Pro effective today?"),    ("t12", "c4",        "SSO login is broken and nobody on our team can get in. This "        "is urgent."),    ("t13", "c3",        "Do you offer a student discount?"),    ("t14", "c1",        "I want to cancel and get money back for the part of the "        "month I won't use."),    ("t15", "c5",        "I cancelled 45 days ago and want my account back with my "        "data."),    ("t16", "c8",        "How do I change the billing email on my account?"),    ("t17", "c10",        "I'm on the annual plan at $39 a month but the invoice says "        "$49. Please fix it."),    ("t18", "c6",        "I was charged yesterday by mistake. Can I get that charge "        "back?"),    ("t19", "c9",        "Our card is updated now. Can you retry the charge today "        "instead of waiting?"),    ("t20", "c3",        "Thanks for the CSV help. Is there a row limit on the export?"),]GOLD = {  # (priority, needs_human), written before any model was run    "t1": ("medium", True),    "t2": ("high", False),    "t3": ("low", False),    "t4": ("medium", True),    "t5": ("medium", False),    "t6": ("medium", True),    "t7": ("high", True),    "t8": ("low", False),    "t9": ("high", True),    "t10": ("high", True),    "t11": ("medium", False),    "t12": ("high", True),    "t13": ("low", True),    "t14": ("medium", True),    "t15": ("low", True),    "t16": ("low", True),    "t17": ("medium", True),    "t18": ("medium", False),    "t19": ("high", True),    "t20": ("low", True),}def main():    client = anthropic.Anthropic()    right = 0    cost = 0.0    for tid, customer, text in TICKETS:        verdict, route, c = triage_with_cascade(            client, {"customer_id": customer, "text": text}        )        priority, human = GOLD[tid]        match = verdict["priority"] == priority and verdict["needs_human"] == human        right += match        cost += c        print(f"{tid} route={route} both_right={match}")    print(f"both fields right {right}/{len(TICKETS)}, "          f"cost per ticket ${cost / len(TICKETS):.5f}")if __name__ == "__main__":    main()