AI agent token budget enforcement in Python: code patterns for hard spending limits

import tiktoken
from dataclasses import dataclass, field
from typing import Optional

COST_PER_1K = {
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "gpt-4o-mini": {"input": 0.000150, "output": 0.000600},
    "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
}

class BudgetExceededException(Exception):
    def __init__(self, session_id: str, budget: float, spent: float, estimated: float):
        self.session_id = session_id
        self.budget = budget
        self.spent = spent
        self.estimated = estimated
        super().__init__(
            f"Session {session_id}: budget ${budget:.4f}, "
            f"spent ${spent:.4f}, estimated call ${estimated:.4f} — would exceed budget"
        )

@dataclass
class BudgetTracker:
    session_id: str
    budget_usd: float
    model: str = "gpt-4o-mini"
    output_estimate_factor: float = 0.7
    _spent_usd: float = field(default=0.0, init=False)

    def estimate_call_cost(self, messages: list[dict], max_tokens: int = 1024) -> float:
        """Estimate cost of an API call before making it."""
        enc = tiktoken.encoding_for_model(self.model)
        input_tokens = sum(
            len(enc.encode(msg.get("content", "") or "")) + 4
            for msg in messages
        ) + 3  # per-message overhead

        estimated_output = int(max_tokens * self.output_estimate_factor)
        rates = COST_PER_1K.get(self.model, COST_PER_1K["gpt-4o-mini"])
        return (
            input_tokens / 1000 * rates["input"] +
            estimated_output / 1000 * rates["output"]
        )

    def check_budget(self, messages: list[dict], max_tokens: int = 1024) -> float:
        """
        Check if the budget allows this call. Returns estimated cost.
        Raises BudgetExceededException if the call would exceed the budget.
        """
        estimated = self.estimate_call_cost(messages, max_tokens)
        if self._spent_usd + estimated > self.budget_usd:
            raise BudgetExceededException(
                self.session_id, self.budget_usd, self._spent_usd, estimated
            )
        return estimated

    def record_actual_cost(self, input_tokens: int, output_tokens: int) -> None:
        """Record the actual cost after a successful API call."""
        rates = COST_PER_1K.get(self.model, COST_PER_1K["gpt-4o-mini"])
        actual = input_tokens / 1000 * rates["input"] + output_tokens / 1000 * rates["output"]
        self._spent_usd += actual

    @property
    def remaining_budget(self) -> float:
        return max(0.0, self.budget_usd - self._spent_usd)

    @property
    def spent(self) -> float:
        return self._spent_usd

import threading
import time
from contextlib import contextmanager
from typing import Optional
import redis

class MultiScopeBudgetEnforcer:
    """
    Thread-safe budget enforcement at session, user, and plan scopes.
    Uses Redis atomic INCRBYFLOAT for distributed accumulation.
    """

    def __init__(self, redis_client: redis.Redis, budgets: dict):
        self.redis = redis_client
        self.budgets = budgets  # {"session": 1.0, "user_daily": 5.0, "plan_monthly": 20.0}
        self._local = threading.local()

    def _scope_key(self, scope: str, identifier: str) -> str:
        if scope == "user_daily":
            day = time.strftime("%Y-%m-%d")
            return f"budget:user_daily:{identifier}:{day}"
        elif scope == "plan_monthly":
            month = time.strftime("%Y-%m")
            return f"budget:plan_monthly:{identifier}:{month}"
        else:
            return f"budget:session:{identifier}"

    def get_spent(self, scope: str, identifier: str) -> float:
        key = self._scope_key(scope, identifier)
        val = self.redis.get(key)
        return float(val) if val else 0.0

    def check_and_reserve(
        self,
        session_id: str,
        user_id: str,
        plan_id: str,
        estimated_cost: float
    ) -> None:
        """
        Atomically check all budget scopes and reserve the estimated cost.
        Raises BudgetExceededException if any scope would be exceeded.
        Uses a Lua script for atomicity.
        """
        lua_script = """
        local keys = KEYS
        local estimated = tonumber(ARGV[1])
        local budgets = {}
        for i = 2, #ARGV do
            budgets[i-1] = tonumber(ARGV[i])
        end
        -- Check all scopes
        for i, key in ipairs(keys) do
            local current = tonumber(redis.call('GET', key) or 0)
            if current + estimated > budgets[i] then
                return {i, current, budgets[i]}
            end
        end
        -- All checks passed: increment all scopes
        for i, key in ipairs(keys) do
            redis.call('INCRBYFLOAT', key, estimated)
            redis.call('EXPIRE', key, 2678400)  -- 31-day TTL
        end
        return nil
        """
        keys = [
            self._scope_key("session", session_id),
            self._scope_key("user_daily", user_id),
            self._scope_key("plan_monthly", plan_id),
        ]
        budget_values = [
            self.budgets["session"],
            self.budgets["user_daily"],
            self.budgets["plan_monthly"],
        ]

        result = self.redis.eval(
            lua_script,
            len(keys),
            *keys,
            str(estimated_cost),
            *[str(b) for b in budget_values],
        )

        if result is not None:
            scope_idx, current, budget = int(result[0]) - 1, float(result[1]), float(result[2])
            scope_names = ["session", "user_daily", "plan_monthly"]
            raise BudgetExceededException(
                session_id=session_id,
                budget=budget,
                spent=current,
                estimated=estimated_cost,
            )

from typing import Callable, Optional, Any
from enum import Enum

class DegradationStrategy(Enum):
    HARD_STOP = "hard_stop"
    FALLBACK_RESPONSE = "fallback_response"
    DOWNGRADE_MODEL = "downgrade_model"
    SUMMARIZE_AND_CONTINUE = "summarize_and_continue"

class BudgetedAgent:
    def __init__(
        self,
        tracker: BudgetTracker,
        on_budget_exceeded: Optional[Callable] = None,
        strategy: DegradationStrategy = DegradationStrategy.FALLBACK_RESPONSE,
        fallback_message: str = "I’ve reached the usage limit for this session. Please start a new conversation or upgrade your plan.",
        downgrade_model: Optional[str] = None,
    ):
        self.tracker = tracker
        self.strategy = strategy
        self.fallback_message = fallback_message
        self.downgrade_model = downgrade_model
        self.on_budget_exceeded = on_budget_exceeded or self._default_handler

    def _default_handler(self, exc: BudgetExceededException) -> Any:
        if self.strategy == DegradationStrategy.HARD_STOP:
            raise exc
        elif self.strategy == DegradationStrategy.FALLBACK_RESPONSE:
            return {"role": "assistant", "content": self.fallback_message}
        elif self.strategy == DegradationStrategy.DOWNGRADE_MODEL:
            return {"downgrade_to": self.downgrade_model, "retry": True}
        else:
            return {"role": "assistant", "content": self.fallback_message}

    def call(self, client, messages: list[dict], max_tokens: int = 512, **kwargs) -> Any:
        try:
            self.tracker.check_budget(messages, max_tokens)
        except BudgetExceededException as exc:
            return self.on_budget_exceeded(exc)

        response = client.chat.completions.create(
            model=self.tracker.model,
            messages=messages,
            max_tokens=max_tokens,
            **kwargs
        )
        usage = response.usage
        self.tracker.record_actual_cost(usage.prompt_tokens, usage.completion_tokens)
        return response.choices[0].message

import runguard
from openai import OpenAI

# Wrap your OpenAI client with RunGuard enforcement
openai_client = OpenAI()
guarded_client = runguard.wrap(
    openai_client,
    session_budget_usd=1.00,
    user_daily_budget_usd=5.00,
    on_exceeded="fallback",  # or "raise", "downgrade"
    fallback_message="Usage limit reached for this session.",
    metadata={"user_id": user_id, "feature": "research_agent"}
)

# Use exactly like the standard OpenAI client — enforcement is transparent
response = guarded_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    max_tokens=512
)