Workflows, not agents: building AI features that spend money

Orbit is an assistant I built that makes Instagram posts for local shops. Nearly everything it does either costs money, because every image is a paid model call, or happens in public, because a post goes out under a shop's name. That changes how it should be built. This is the case for a plain workflow over an agent loop, and for the unglamorous engineering that makes one dependable.

Agents are for steps you can't predict

An agent loop hands the model some tools and lets it choose: call one, read the result, choose again, stop when it decides it's done. That's the right shape when the steps can't be known ahead of time, like debugging an unfamiliar codebase.

A festival post for a shop is not like that. It's the same sequence every time:

Wrapping that in a loop would buy non-determinism and pay for it in tokens, latency and debugging time. The stakes point the same way: a loop that can decide to call the image tool four times can quietly spend four times as much, and post something nobody approved.

The model still does real work. It writes the concept, judges the artwork and drafts the caption. It just doesn't decide what happens next.

A bounded retry is not an agent

The pipeline repeats itself in exactly one place. After the artwork is painted, a second model call acts as an art director and lists what's wrong with it. If it finds problems, the artwork is painted once more with those problems named in the prompt.

The model says what's wrong; code decides whether there's another attempt, and there is at most one. Each extra step must also fit a deadline. The request is cut off at 60 seconds, so the pipeline works to a 50-second budget: it only runs the check if 9 seconds remain, and only repaints if 32 do. If the image model is down entirely, the post is made with the owner's own photo, or with the words alone. That fallback is a branch in code, not something a model has to think of.

One gateway, and workers that say what they're for

Orbit didn't start this way. Three services each built their own SDK client, four more classes inherited from the caption service just to borrow one, and eight call sites repeated the same parse-validate-raise steps. None of them read the token usage the API returns, so nobody could say what anything cost.

Now one gateway does all of it. A caller passes a JSON schema and a prompt and gets back a validated Pydantic model. Every failure becomes a user-facing error once, in that file, and every call, successful or not, writes one row to a cost ledger. Here it is, simplified from Orbit's Python and FastAPI backend:

import json
import time
 
from pydantic import BaseModel, ValidationError
 
 
class AIError(Exception):
    def __init__(self, status: int, message: str) -> None:
        super().__init__(message)
        self.status = status  # the message is safe to show a shop owner
 
 
class ProviderBusy(Exception): ...  # still a 429 after the SDK's retries
class ProviderDown(Exception): ...  # a 5xx or a dropped connection
 
 
class Gateway:
    """The only object in the codebase that calls the model SDK."""
 
    def __init__(self, client, ledger, *, purpose: str, model: str) -> None:
        self.client, self.ledger, self.purpose, self.model = client, ledger, purpose, model
 
    async def structured(self, *, schema: dict, prompt: str, into: type[BaseModel]) -> BaseModel:
        started = time.monotonic()
        try:
            reply = await self.client.create(model=self.model, prompt=prompt, json_schema=schema)
        except ProviderBusy as exc:
            await self.record(None, started, "rate_limited")
            raise AIError(503, "The AI is busy right now — try again in a moment.") from exc
        except ProviderDown as exc:
            await self.record(None, started, "provider_error")
            raise AIError(502, "The AI didn't answer. Try again.") from exc
 
        outcome = {"refusal": "refusal", "max_tokens": "truncated"}.get(reply.stop_reason, "ok")
        await self.record(reply, started, outcome)  # exactly one row per call
        if reply.stop_reason == "refusal":
            raise AIError(422, "The AI declined to generate content for this request.")
        if reply.stop_reason == "max_tokens":
            raise AIError(502, "AI response was cut off. Try again.")
        try:
            return into.model_validate(json.loads(reply.text))
        except (json.JSONDecodeError, ValidationError) as exc:
            raise AIError(502, "AI returned an unparseable response. Try again.") from exc
 
    async def record(self, reply, started: float, outcome: str) -> None:
        usage = getattr(reply, "usage", None)
        await self.ledger.record(
            purpose=self.purpose,
            model=self.model,
            outcome=outcome,
            input_tokens=getattr(usage, "input_tokens", 0),
            output_tokens=getattr(usage, "output_tokens", 0),
            latency_ms=int((time.monotonic() - started) * 1000),
        )

On top sit workers. A Worker is one job, one purpose, one gateway, and its only contract is a PURPOSE string that names the work in the ledger. The cost of captions versus art briefs versus reply drafts becomes a query, not a guess.

from typing import ClassVar
 
from gateway import Gateway
 
 
class Worker:
    """One job, one purpose, one gateway."""
 
    PURPOSE: ClassVar[str] = "unknown"  # every subclass sets it
 
    def __init__(self, client, ledger, model: str) -> None:
        self._ai = Gateway(client, ledger, purpose=self.PURPOSE, model=model)
 
 
class CaptionWriter(Worker):
    PURPOSE = "caption"

The base is deliberately thin, with no shared run(), prompt or schema, because Orbit's twenty workers differ in shape: some return a model, some a dict, one falls back to a default instead of raising. One signature for all of them would cost more in special cases than it saves.

Rules live in a test, not in someone's head

Layering only helps if nobody routes around it. A second SDK client with no ledger row is invisible in code review and shows up later as a bill nobody can explain. So the rules are tests that read the source:

import ast
import re
from pathlib import Path
 
APP = Path(__file__).resolve().parents[1] / "app"
SDK_CALL = re.compile(r"ModelClient\(|\.messages\.create\(")
 
 
def test_only_the_gateway_touches_the_sdk():
    offenders = [
        f for f in APP.rglob("*.py")
        if f.name != "gateway.py" and SDK_CALL.search(f.read_text())
    ]
    assert offenders == [], f"No ledger row, no error mapping: {offenders}"
 
 
def test_every_worker_names_itself():
    unnamed = []
    for path in APP.rglob("*.py"):
        for node in ast.walk(ast.parse(path.read_text())):
            if not isinstance(node, ast.ClassDef):
                continue
            if not any(isinstance(b, ast.Name) and b.id == "Worker" for b in node.bases):
                continue
            if not any(
                isinstance(s, ast.Assign) and any(getattr(t, "id", "") == "PURPOSE" for t in s.targets)
                for s in node.body
            ):
                unnamed.append(f"{path.name}::{node.name}")
    assert unnamed == [], f"These would bill to 'unknown': {unnamed}"

A third test checks that only worker.py constructs a gateway, so no spend lands in the ledger under a hand-typed purpose.

file allowed to call the model SDK

1

file allowed to call the model SDK

file allowed to build a gateway

1

file allowed to build a gateway

workers, each naming its PURPOSE

20

workers, each naming its PURPOSE

A receipt for every call

Each ledger row holds the model, purpose, workspace, input and output tokens, cache reads and writes, latency, outcome and cost. Three decisions made it worth having.

Money is an integer. Cost is stored as integer micro-dollars in a 64-bit column, and the arithmetic is neat: tokens multiplied by dollars per million tokens is already micro-dollars. At an illustrative $3 per million, 1,200 input tokens cost 3,600 micro-dollars. Each call is rounded once, and every sum after that is exact.

An unknown model is expensive, not free. Rates live in one table, but models get renamed and dated snapshot names appear. A lookup that falls back to zero would make a new, expensive model look free, which is the one thing the ledger exists to prevent.

RATES = {"text-large": (5.00, 25.00), "text-medium": (3.00, 15.00)}  # $/MTok in, out (illustrative)
DEFAULT = RATES["text-medium"]
 
 
def cost_micros(model: str, input_tokens: int, output_tokens: int) -> int:
    # The longest known name inside the id, so "text-large-2026-06-01" still
    # matches; otherwise the default. Never zero.
    known = next((k for k in sorted(RATES, key=len, reverse=True) if k in model), None)
    rate_in, rate_out = RATES[known] if known else DEFAULT
    return round(input_tokens * rate_in + output_tokens * rate_out)
 
 
assert cost_micros("text-medium", 1_200, 0) == 3_600

Text models fall back to the default model's rate. Images and video fall back to the most expensive model I use. Overstating a cost prompts a look; understating it hides a leak.

Failures are recorded, and metering never breaks the work. Refusals, truncated answers and provider errors all get rows, because a run of failures belongs next to the spend. For images, where the provider only bills for a picture that came back, a failed attempt is recorded at zero cost. And the ledger write is best effort: if it can't reach the database, it logs a warning and the call carries on. A missing row costs visibility; a raised exception would cost someone their post.

Retries, and errors a shop owner can read

The text model's SDK already retries rate limits, server errors and dropped connections with backoff. The gateway sets max_retries=2 and adds nothing on top, because a second retry layer would multiply the waiting. One subtlety: the timeout goes through the SDK's per-request options so it applies to each attempt. Wrapping the call in asyncio.timeout would cap the whole retry sequence instead.

The image provider is called over plain HTTP, so it gets its own loop:

import asyncio
 
import httpx
 
RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})
BACKOFF_SECONDS = (2.0, 6.0)
 
 
async def post_with_retries(client: httpx.AsyncClient, url: str, payload: dict) -> httpx.Response:
    response = await client.post(url, json=payload)
    for delay in BACKOFF_SECONDS:
        if response.status_code not in RETRY_STATUSES:
            break
        await asyncio.sleep(delay)
        response = await client.post(url, json=payload)
    return response

The first version retried only 429. In practice the common failure was a 503 from an overloaded model, which got no retry at all, though it needs exactly the same treatment.

The provider's own error text goes to the logs, never to the screen. Each failure is mapped once:

Idempotency keys on anything that spends

A phone on a train times out and the app tries again. An owner double-clicks. Without protection, each retry pays again: a second image, a second video, a second slice of the month's allowance.

So the endpoints that spend money accept an Idempotency-Key header. Send the same key twice and the second request gets the first one's answer. The details are where it gets interesting:

• Opt-in. No header, no change in behaviour.
• Scoped to the workspace. Two shops can both send retry-1.
• Body fingerprinted with sha256. The same key with a different body is a 422, because replaying would answer a question the caller didn't ask.
• In flight means 409, with Retry-After: 5.
• Failure releases the key, or every retry would get a 409 at exactly the moment the client most wants to retry.
• One atomic claim: INSERT ... ON CONFLICT DO NOTHING RETURNING, so two racing requests can't both win.

import hashlib
 
from fastapi import HTTPException, Request
 
 
async def claim(request: Request, store, workspace_id: str, endpoint: str):
    """Returns (key, replay). key is None when the request runs unguarded."""
    key = request.headers.get("Idempotency-Key")
    if not key:
        return None, None  # opt-in
    fingerprint = hashlib.sha256(await request.body()).hexdigest()
    try:
        if await store.insert_if_absent(workspace_id, key, endpoint, fingerprint):
            return key, None  # we own it: do the work, then store the result
        row = await store.get(workspace_id, key)
    except Exception:
        return None, None  # fail open
    if row is None:
        return key, None  # swept between the insert and the read
    if (row.endpoint, row.fingerprint) != (endpoint, fingerprint):
        raise HTTPException(422, "This Idempotency-Key was already used for a different request.")
    if row.status == "running":
        raise HTTPException(409, "That request is still running.", headers={"Retry-After": "5"})
    return key, row.response  # replay the first answer

A context manager around it deletes a claim that is still running if the handler raises. In Orbit it's a decorator on nine endpoints, from captions to video, placed under the router decorator with functools.wraps so FastAPI still reads the handler's real signature. Results are kept for 24 hours.

Rate limits that give back what they took

Every generation costs real money, so the expensive endpoints are rate-limited: a fixed window, counted in Postgres with an upsert. Two details matter more than the limit itself.

Count late. The limiter is a plain function rather than a FastAPI dependency, because a dependency counts every attempt. Called after validation and after the cache, it only counts work that costs something. A typo, or asking again for something already made, is free.

Refund what was my fault. The count is spent before the expensive call, so nobody can start ten at once. If the call then fails on Orbit's side, the count is given back. Bad input is still charged.

from fastapi import HTTPException, Request
 
# enforce_rate_limit and release_rate_limit are the limiter's two functions;
# validate, cached_result and generate stand in for the real steps.
 
 
async def generate_post(request: Request, payload: dict):
    spec = validate(payload)  # bad input fails here, uncounted
    if cached := await cached_result(spec):  # asking twice is free
        return cached
    await enforce_rate_limit(request, "generate", limit=3, window_seconds=3600)
    try:
        return await generate(spec)
    except HTTPException as exc:
        if exc.status_code >= 500:  # our provider failed, not their input
            await release_rate_limit(request, "generate", window_seconds=3600)
        raise

The refund is one guarded update that can never go below zero:

UPDATE counters SET count = count - 1
WHERE key = :key AND window_start = :window_start AND count > 0;

The limiter fails open too: an outage in the counter must not become an outage in the product. Signed-in features follow the same principle. For video, the monthly limit is checked before the provider is called, but usage is recorded only after a video comes back.

Variety that stays put

The first festival posts all looked alike: the same greeting in the same font, centred at the same height. So I added seven "treatments". Each sets the headline's position, alignment, size, case, letter spacing and typeface, where the logo goes, and a "calm region" the image model is told to keep quiet. A shop's trade picks a mood (a café is warm, a jeweller luxury), and the shop's name picks a treatment within it.

The choice must be deterministic both ways. Two shops should look different, and one shop should look the same every time, so next festival's post is recognisably theirs. Randomness would give variety and lose the brand. And because the treatment is chosen before the artwork, the image prompt can ask for calm space exactly where the words will land.

Why not hash()

The obvious hash(name) % len(options) is wrong. Python salts string hashing per process:

$ for i in 1 2 3 4; do python3 -c "print(hash(\"Asha's Bakery\") % 7)"; done
3
5
5
1

On a server, every shop would get a new look after every deploy. sha256 is the same on every machine, every time. The first version of the picker looked like this:

import hashlib
 
 
def treatment_for(shop_name: str, options: tuple[str, ...]) -> str:
    key = (shop_name or "").strip().lower() or "orbit"
    digest = hashlib.sha256(key.encode("utf-8")).digest()
    return options[int.from_bytes(digest[:8], "big") % len(options)]

The list was a contract

That first version took the digest modulo the number of options, which made the order of the tuple part of the contract: reorder it, or remove an entry, and existing shops change layout. So it carried a comment: append only.

While writing this post I tested that rule, and my own comment was too generous. It said appending was safe. With a modulo pick it isn't, because the divisor changes. I ran 20,000 made-up shop names through the real function: appending one treatment that suits cafés moved about 80% of cafés to a different layout, most of them to one of the old layouts rather than the new one.

So Orbit now picks with rendezvous hashing: score every option against the shop, by name, and take the highest.

import hashlib
 
 
def score(key: str, name: str) -> int:
    digest = hashlib.sha256(f"{key}\x1f{name}".encode()).digest()
    return int.from_bytes(digest[:8], "big")
 
 
def rank(key: str, names: list[str]) -> list[str]:
    """Every option, the best match for this shop first."""
    return sorted(set(names), key=lambda name: (-score(key, name), name))

Adding a fifth option to four now moves about one name in five (19.7% in the same 20,000-name run), and every one of them moves to the new option. Removing or turning off an option moves only the shops that had it, and the order of the list stops mattering. Festival layouts use the same ranking: concept one takes the shop's favourite layout, concept two its second favourite, so three concepts still get three different layouts.

Each of those properties has a test, and the tests fail when pointed at the old modulo version. The switch itself moved existing shops once. Nothing stored reads the choice back, so posts already made keep their look; only the next ones can change.

Where a loop earns its place

In one feature the model's output does shape what happens next. Ask Orbit takes a sentence from the owner, like the example in the code: "Create a post for Gandhi Jayanti using our logo colours and publish on Instagram and Facebook on Oct 2nd at 10am."

The model only extracts. It fills a schema (occasion, colours, channels, local time, a note) and is told to ask a short question rather than guess. A pure function then reconciles that with reality: the festival calendar, the channels actually connected, and the owner's timezone from their browser. A time that has passed becomes a question; a channel that isn't connected becomes a warning, not a blocker.

The planning endpoint generates and schedules nothing. Only after the owner confirms does anything get spent or published. That confirm step isn't a safety feature bolted on; it is the feature, and it's why a loop is acceptable here.

What I'd do differently

• Build the gateway before the second model call. By the time Orbit had one, eight call sites had their own error handling, and the most expensive calls, images and video, joined the ledger later still. Every week without a ledger is a week of costs you can only estimate.
• Pick with rendezvous hashing from day one. Switching later moved every existing shop once. "Append only" is a rule people must remember; a function that ignores order is one they don't.
• Use the cache columns. The ledger has tracked cache reads and writes from the start, but Orbit sends no prompt-caching hints yet. Its system prompts are long and stable, which is exactly where caching pays.

Agents are for problems whose steps you can't write down. When you can, and each step costs money or happens in public, write them down, and spend your effort on the plumbing around the model.