How to Control LLM API Costs: Token Budgets, Batching and a Budget Guard in Python

Most surprises on an LLM bill do not come from the price per token. They come from the call pattern: how many requests you send, how much context each one carries, and whether anything stops a loop that runs too long.

In this tutorial you will build a small budget guard in Python, learn how to count tokens before you send a request, and see why batching often matters more than choosing a cheaper model. The examples use a generic client, so you can adapt them to any provider.

Cost is driven by the call pattern

Provider pricing is simple: input tokens times the input price, plus output tokens times the output price. The bill is large when one of these grows without a limit:

• Number of calls (a loop over thousands of items).
• Input size per call (the full document sent each time).
• Output size (no maximum set).
• Retries (a failing call repeated automatically).

I learned this in a real project. A script called a paid data API once per keyword, 1,984 single requests in one day, and the day cost 23.92 USD. The same data was available through a batch endpoint that accepts up to 700 items per call. Batched, the cost would have been roughly 2 to 3 USD. The price per item never changed. Only the call pattern did.

LLM APIs behave the same way, so apply the same thinking before you write the loop.

Step 1: Count tokens before you send

You can estimate the cost of a request before you make it. Many providers publish a tokenizer or a token counting endpoint. If yours does not, a rough rule for English text is about four characters per token. It is an estimate, so keep a safety margin.

def estimate_tokens(text: str) -> int:
    # Rough estimate for English text. Replace with your provider's tokenizer.
    return max(1, len(text) // 4)

Use the real tokenizer when accuracy matters. Use the estimate when you only need to catch a request that is ten times bigger than expected.

Step 2: A budget guard class

The guard keeps a running total and refuses to continue when the budget is used up. It raises an exception, so a runaway loop stops instead of billing you all night.

from dataclasses import dataclass, field


class BudgetExceeded(Exception):
    pass


@dataclass
class BudgetGuard:
    limit_usd: float
    input_price_per_1m: float
    output_price_per_1m: float
    spent_usd: float = field(default=0.0, init=False)

    def check(self, input_tokens: int, max_output_tokens: int) -> None:
        worst_case = self._cost(input_tokens, max_output_tokens)
        if self.spent_usd + worst_case > self.limit_usd:
            raise BudgetExceeded(
                f"Budget {self.limit_usd:.2f} USD would be exceeded "
                f"(spent {self.spent_usd:.2f}, next call up to {worst_case:.2f})"
            )

    def record(self, input_tokens: int, output_tokens: int) -> None:
        self.spent_usd += self._cost(input_tokens, output_tokens)

    def _cost(self, input_tokens: int, output_tokens: int) -> float:
        return (
            input_tokens * self.input_price_per_1m
            + output_tokens * self.output_price_per_1m
        ) / 1_000_000

The prices are constructor arguments on purpose. Prices change, and a constant buried in the code goes stale. Read them from your provider's current price list.

Use it around every call:

guard = BudgetGuard(limit_usd=5.0, input_price_per_1m=3.0, output_price_per_1m=15.0)

def ask(client, prompt: str) -> str:
    max_out = 500
    guard.check(estimate_tokens(prompt), max_out)
    response = client.complete(prompt, max_tokens=max_out)  # your provider call
    guard.record(response.input_tokens, response.output_tokens)
    return response.text

Two design choices are worth noting. The check uses the worst case (full max_tokens output), so the guard errs on the safe side. And the exception is loud: you want a stopped job and a log line, not a silent fallback.

Step 3: Batching, collect first and send once

The most common anti-pattern looks like this:

for item in items:                 # 2,000 items
    result = ask(client, f"Classify: {item}")

Each call repeats the same instructions, adds network overhead, and multiplies the chance of hitting a rate limit. When the task allows it, collect first and send once:

def chunks(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]

for batch in chunks(items, 50):
    numbered = "\n".join(f"{n}. {text}" for n, text in enumerate(batch, 1))
    prompt = (
        "Classify each numbered item as positive, neutral or negative. "
        "Answer with one line per item in the format '<number>: <label>'.\n\n"
        + numbered
    )
    answer = ask(client, prompt)

Now 2,000 items need 40 calls instead of 2,000, and the instructions are sent 40 times instead of 2,000 times. Test the batch size on a sample: very large batches can reduce answer quality, and a parsing step must handle a missing line.

Many providers also offer an asynchronous batch interface at a reduced price for jobs that can wait. If your job is not interactive, check for it.

Step 4: Avoid input bloat

The second big multiplier is context. If you send the entire document on every call and only one paragraph changed, you pay for the unchanged part again and again.

Practical rules:

• Send only the part that the model needs for this question.
• Keep the fixed instructions short, and remove examples that do not change the answer.
• For chat-style flows, summarize old turns instead of replaying them in full.
• If your provider supports prompt caching, put the stable part at the start of the prompt and the variable part at the end, then check the pricing rules for cached input.

Step 5: Monitoring and daily caps

A guard inside the script protects one job. You also need a limit across jobs:

1. Set a hard monthly or daily spending limit in your provider's dashboard, if it offers one.
2. Log every call with timestamp, job name, input tokens and output tokens. A CSV file is enough to start.
3. Review the log once a week and look for the job whose call count jumped.

import csv, time

def log_call(job: str, input_tokens: int, output_tokens: int, path="llm_calls.csv"):
    with open(path, "a", newline="") as f:
        csv.writer(f).writerow([time.strftime("%Y-%m-%d %H:%M:%S"), job, input_tokens, output_tokens])

Call log_call right after guard.record. The first time a loop goes wrong, this file tells you which job caused it.

Summary

• Cost follows the call pattern: number of calls, input size, output limit, retries.
• Estimate tokens before you send, and stop with an exception when the budget is gone.
• Collect items and send them in batches instead of one request per item.
• Send only the context the model needs.
• Log every call and set a cap outside your code as well.

None of this needs a special library. It needs the habit of asking, before every loop, "How many calls will this make, and what happens if it runs ten times longer than expected?"