6. Cost Control and Rate Limiting

Overview

In the previous capsule you implemented error recovery and resilience: graceful degradation, fallbacks, circuit breakers, retry policies. Your agent survives failures. But there's a silent failure that doesn't raise exceptions or break requests — uncontrolled cost. An AI agent isn't a CRUD endpoint that costs fractions of a cent per request. A research query can involve 5-15 LLM calls, each consuming thousands of tokens, plus tool calls to paid APIs, plus embedding generation for memory. Without cost control, an agent in production can burn a whole month's budget in hours.

The numbers are real: a complex research query to the Research Agent can consume 15,000-30,000 tokens across planning, tool calls, synthesis, and reflection. With GPT-4o at $2.50/M input and $10.00/M output, a single query can cost between $0.10 and $0.50. Multiply by 500 daily queries and you're spending $50-$250 a day — $1,500-$7,500 a month. And that's just one agent. If you have a multi-agent system where the supervisor delegates to 3 sub-agents, multiply by 3-4x.

Connection to the module: In capsule 04 you configured monitoring and cost metrics. In this capsule, you go from measuring costs to actively controlling them. Token budgets that abort expensive requests before they ruin you. Rate limiting that protects against abuse. Model routing that sends cheap tasks to cheap models. Caching that eliminates redundant calls. Cost awareness isn't a nice-to-have — it's the AI Engineer's responsibility.

Anatomy of an Agent's Cost

Where the tokens go

To control costs, you first need to understand where every cent is spent. An agent doesn't make a single LLM call — it makes many, and each has a different cost:

Cost anatomy: Research Agent — a typical query
═══════════════════════════════════════════════════════════════

Phase              Input Tokens  Output Tokens  Model       Cost
──────────────────────────────────────────────────────────────────
Planning            2,200          800          gpt-4o      $0.014
Tool: web_search       —            —           Tavily      $0.001
Tool: web_search       —            —           Tavily      $0.001
Synthesis           8,500         1,200         gpt-4o      $0.033
Reflection          4,000          600          gpt-4o      $0.016
Final answer        3,800         1,500         gpt-4o      $0.025
Checkpoint write       —            —           PostgreSQL  $0.000
──────────────────────────────────────────────────────────────────
TOTAL              18,500         4,100                     $0.090

With multi-agent (supervisor + 3 sub-agents):
──────────────────────────────────────────────────────────────────
Supervisor routing   1,500          300          gpt-4o      $0.007
Sub-agent 1         12,000         2,800         gpt-4o      $0.058
Sub-agent 2          8,000         1,500         gpt-4o      $0.035
Sub-agent 3          6,500         1,200         gpt-4o      $0.029
Final merge          5,000         2,000         gpt-4o      $0.033
──────────────────────────────────────────────────────────────────
TOTAL              33,000         7,800                     $0.162

The hidden multipliers

What you see in the prompt isn't all you pay for. There are hidden costs that add up:

Calculating the real cost

from dataclasses import dataclass, field

MODEL_PRICING = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
    "gpt-4.1": {"input": 2.00 / 1_000_000, "output": 8.00 / 1_000_000},
    "gpt-4.1-mini": {"input": 0.40 / 1_000_000, "output": 1.60 / 1_000_000},
    "gpt-4.1-nano": {"input": 0.10 / 1_000_000, "output": 0.40 / 1_000_000},
}


@dataclass
class CostTracker:
    entries: list[dict] = field(default_factory=list)

    def record(self, model: str, input_tokens: int, output_tokens: int,
               component: str = "unknown") -> float:
        pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-4o-mini"])
        cost = (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
        self.entries.append({
            "model": model, "input_tokens": input_tokens,
            "output_tokens": output_tokens, "cost_usd": cost,
            "component": component,
        })
        return cost

    def total_cost(self) -> float:
        return sum(e["cost_usd"] for e in self.entries)

    def cost_by_component(self) -> dict[str, float]:
        breakdown: dict[str, float] = {}
        for e in self.entries:
            breakdown[e["component"]] = breakdown.get(e["component"], 0) + e["cost_usd"]
        return breakdown

If 70% of the cost is in reflection, you know where to optimize.

Token Budgets per Request

The concept

A token budget is a hard limit: "this request can't consume more than X tokens." If the agent gets close to the limit, it aborts the current iteration and generates an answer with what it has. Without budgets, an agent with a reflection loop can iterate indefinitely, burning tokens each time.

Implementation with LangGraph

Add token_budget and tokens_consumed to the agent's state. Before each LLM call, check whether there's budget left. If not, generate a partial answer with what it has:

from typing import TypedDict
from langchain_core.messages import AIMessage


class AgentState(TypedDict):
    messages: list
    token_budget: int
    tokens_consumed: int
    budget_exceeded: bool


def check_budget(state: AgentState) -> AgentState:
    ratio = state["tokens_consumed"] / state["token_budget"] if state["token_budget"] > 0 else 0
    if ratio >= 0.95:
        state["budget_exceeded"] = True
    return state


def should_continue(state: AgentState) -> str:
    if state.get("budget_exceeded"):
        return "generate_partial_response"
    return "continue_processing"


def generate_partial_response(state: AgentState) -> AgentState:
    warning = (
        f"Token budget reached ({state['tokens_consumed']:,} / "
        f"{state['token_budget']:,}). Answering with partial information."
    )
    state["messages"].append(AIMessage(content=warning))
    return state

A LangChain callback for automatic tracking

Instead of tracking manually, use a callback handler that intercepts every LLM call:

from langchain_core.callbacks import BaseCallbackHandler


class BudgetCallbackHandler(BaseCallbackHandler):
    def __init__(self, max_tokens: int = 50_000):
        self.max_tokens = max_tokens
        self.total_consumed = 0
        self.calls: list[dict] = []

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        total = input_tokens + output_tokens
        self.total_consumed += total
        self.calls.append({
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total": total,
            "accumulated": self.total_consumed,
        })

        if self.total_consumed > self.max_tokens:
            raise TokenBudgetExceeded(
                f"Budget exceeded: {self.total_consumed:,} / {self.max_tokens:,}"
            )

    @property
    def remaining(self) -> int:
        return max(0, self.max_tokens - self.total_consumed)

    @property
    def utilization(self) -> float:
        return self.total_consumed / self.max_tokens if self.max_tokens > 0 else 0


class TokenBudgetExceeded(Exception):
    pass

Use it when invoking the agent:

budget_handler = BudgetCallbackHandler(max_tokens=50_000)

try:
    result = await graph_app.ainvoke(
        {"messages": [HumanMessage(content=query)]},
        config={"callbacks": [budget_handler]},
    )
except TokenBudgetExceeded:
    result = {"messages": [AIMessage(
        content="The token limit for this query was reached. "
        "The answer may be incomplete."
    )]}
finally:
    logger.info(f"Request used {budget_handler.total_consumed:,} tokens "
                f"({budget_handler.utilization:.0%} of budget)")

Rate Limiting per User

Why you need rate limiting

Without rate limiting, a single user (or a script) can exhaust your rate limit with OpenAI, leave everyone else without service, and generate a huge bill. Rate limiting protects your budget, everyone else's experience, and your relationship with the LLM providers.

Implementation with SlowAPI

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from starlette.responses import JSONResponse

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)


@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "error": "rate_limit_exceeded",
            "message": "Too many requests. Please try again later.",
            "retry_after_seconds": 60,
        },
        headers={"Retry-After": "60"},
    )


@app.post("/research")
@limiter.limit("10/minute;100/hour;500/day")
async def research(request: Request, query: str):
    result = await graph_app.ainvoke({"messages": [HumanMessage(content=query)]})
    return {"response": result["messages"][-1].content}

Rate limiting by user plan

A uniform rate limit isn't enough. Differentiate by plan:

from enum import Enum


class UserPlan(Enum):
    FREE = "free"
    PRO = "pro"
    ENTERPRISE = "enterprise"


PLAN_LIMITS = {
    UserPlan.FREE: {
        "requests_per_minute": 3, "requests_per_day": 50,
        "max_tokens_per_request": 20_000, "daily_token_budget": 500_000,
    },
    UserPlan.PRO: {
        "requests_per_minute": 15, "requests_per_day": 1_000,
        "max_tokens_per_request": 50_000, "daily_token_budget": 5_000_000,
    },
    UserPlan.ENTERPRISE: {
        "requests_per_minute": 60, "requests_per_day": 10_000,
        "max_tokens_per_request": 100_000, "daily_token_budget": 50_000_000,
    },
}

A per-user token budget with Redis

Beyond limiting requests, you need to limit the tokens consumed per user. A user could make 3 "legal" requests per minute, each consuming 100,000 tokens:

import redis.asyncio as redis
from datetime import datetime, timezone

redis_client = redis.from_url("redis://localhost:6379")


async def check_user_token_budget(user_id: str, plan: UserPlan) -> dict:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    key = f"token_budget:{user_id}:{today}"
    daily_limit = PLAN_LIMITS[plan]["daily_token_budget"]

    consumed = int(await redis_client.get(key) or 0)
    remaining = daily_limit - consumed

    return {
        "allowed": remaining > 0,
        "consumed_today": consumed,
        "daily_limit": daily_limit,
        "remaining": max(0, remaining),
    }


async def deduct_tokens(user_id: str, tokens: int):
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    key = f"token_budget:{user_id}:{today}"
    pipe = redis_client.pipeline()
    pipe.incrby(key, tokens)
    pipe.expire(key, 86400 * 2)
    await pipe.execute()

The cost control middleware

A middleware that checks the budget before processing each request:

from starlette.middleware.base import BaseHTTPMiddleware


class CostControlMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in ("/health", "/metrics", "/docs"):
            return await call_next(request)

        user_id = request.headers.get("X-User-ID")
        plan = UserPlan(request.headers.get("X-User-Plan", "free"))

        budget_status = await check_user_token_budget(user_id, plan)
        if not budget_status["allowed"]:
            return JSONResponse(status_code=429, content={
                "error": "daily_token_budget_exceeded",
                "consumed": budget_status["consumed_today"],
                "limit": budget_status["daily_limit"],
            })

        request.state.user_plan = plan
        request.state.max_tokens = PLAN_LIMITS[plan]["max_tokens_per_request"]
        return await call_next(request)

Model Routing by Cost

The core idea

Not every task needs the most capable (and most expensive) model. Classifying a query doesn't require GPT-4o — GPT-4o-mini does it just as well at 1/17 of the cost. Planning, classification, parameter extraction: tasks cheap models handle perfectly. Reserve the premium model for synthesis and complex reasoning.

Model Routing: Router (nano) → classifies complexity
├── SIMPLE  → nano  ($0.10/M) — FAQs, greetings, classification
├── MEDIUM  → mini  ($0.40/M) — planning, extraction, summaries
└── COMPLEX → full  ($2.50/M) — synthesis, research, reasoning

Implementing the router

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

router_llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0, max_tokens=20)
cheap_llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)
mid_llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
premium_llm = ChatOpenAI(model="gpt-4.1", temperature=0)

ROUTER_PROMPT = """Classify the task complexity. Respond with exactly one word:
- SIMPLE: greetings, FAQs, yes/no questions, format validation
- MEDIUM: summarization, data extraction, planning, translation
- COMPLEX: multi-step research, analysis, creative writing, reasoning

Task: {query}
Complexity:"""


async def route_by_cost(query: str) -> ChatOpenAI:
    """Selects the cheapest model capable of handling the task."""
    response = await router_llm.ainvoke([
        HumanMessage(content=ROUTER_PROMPT.format(query=query))
    ])
    complexity = response.content.strip().upper()

    model_map = {
        "SIMPLE": cheap_llm,
        "MEDIUM": mid_llm,
        "COMPLEX": premium_llm,
    }
    return model_map.get(complexity, mid_llm)

Routing inside a LangGraph agent

Routing doesn't only apply between requests — it applies inside a single request. The agent's different nodes have different intelligence requirements:

NODE_MODEL_MAP = {
    "classify_query": "gpt-4.1-nano",
    "plan_research": "gpt-4.1-mini",
    "execute_search": "gpt-4.1-mini",
    "synthesize": "gpt-4.1",
    "reflect": "gpt-4.1-mini",
    "generate_response": "gpt-4.1",
}


def get_llm_for_node(node_name: str) -> ChatOpenAI:
    model = NODE_MODEL_MAP.get(node_name, "gpt-4.1-mini")
    return ChatOpenAI(model=model, temperature=0)

The real impact of model routing

The savings aren't marginal — they're transformative:

Model routing cuts the cost in half without degrading the quality of the final answers. Simple tasks come out the same with a cheap model — only synthesis and reasoning need the premium one.

Cost fallback

If the premium model isn't available (rate limit, timeout), you fall back downward — from the assigned model to the next cheapest:

async def invoke_with_cost_fallback(query: str, node_name: str) -> str:
    models = ["gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"]
    primary = NODE_MODEL_MAP.get(node_name, "gpt-4.1-mini")
    start_idx = models.index(primary) if primary in models else 0

    for model_name in models[start_idx:]:
        try:
            llm = ChatOpenAI(model=model_name, temperature=0, request_timeout=30)
            return (await llm.ainvoke([HumanMessage(content=query)])).content
        except Exception:
            continue
    raise RuntimeError("All models failed")

Cost Monitoring

A real-time cost dashboard

In capsule 04 you implemented metrics with Prometheus. Now extend them for complete cost visibility. Your dashboard should show: today's cost vs. the daily limit, monthly cost vs. the budget, a breakdown by component (planning, synthesis, reflection, tools), a breakdown by model, average cost per query (p50, p95, p99), and the top users by spend.

Tracking with Prometheus and Redis

from prometheus_client import Counter, Gauge, Histogram
from datetime import datetime, timezone

cost_total = Counter("agent_cost_usd_total", "Total cost in USD", ["model", "component"])
cost_per_query = Histogram(
    "agent_cost_per_query_usd", "Cost per query",
    buckets=[0.01, 0.02, 0.05, 0.10, 0.20, 0.50, 1.00, 2.00],
)
daily_cost_gauge = Gauge("agent_daily_cost_usd", "Today's total cost")
monthly_budget_usage = Gauge("agent_monthly_budget_pct", "Monthly budget utilization %")

MONTHLY_BUDGET = 750.00


async def record_cost(
    model: str, input_tokens: int, output_tokens: int,
    component: str, user_id: str,
) -> float:
    pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-4o-mini"])
    cost = (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
    cost_total.labels(model=model, component=component).inc(cost)

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    pipe = redis_client.pipeline()
    pipe.incrbyfloat(f"cost:daily:{today}", cost)
    pipe.expire(f"cost:daily:{today}", 86400 * 7)
    pipe.incrbyfloat(f"cost:monthly:{month}", cost)
    pipe.expire(f"cost:monthly:{month}", 86400 * 45)
    pipe.incrbyfloat(f"cost:user:{user_id}:{today}", cost)
    pipe.expire(f"cost:user:{user_id}:{today}", 86400 * 7)
    await pipe.execute()

    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)
    daily_cost_gauge.set(daily)
    monthly_budget_usage.set((monthly / MONTHLY_BUDGET) * 100)
    return cost

Cost alerts

Extend capsule 04's AlertManager with cost rules:

async def check_cost_alerts(query_cost: float):
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)
    budget_pct = (monthly / MONTHLY_BUDGET) * 100

    if query_cost > 0.50:
        logger.warning(f"COST ALERT: Single query cost ${query_cost:.2f}")
    if daily > 100.0:
        logger.critical(f"COST ALERT: Daily cost ${daily:.2f} exceeds $100")
    elif daily > 50.0:
        logger.warning(f"COST ALERT: Daily cost ${daily:.2f} exceeds $50")
    if budget_pct > 90:
        logger.critical(f"COST ALERT: Monthly budget at {budget_pct:.0f}%")
    elif budget_pct > 75:
        logger.warning(f"COST ALERT: Monthly budget at {budget_pct:.0f}%")

A visibility endpoint

@app.get("/costs")
async def get_costs():
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)

    return {
        "daily_cost_usd": round(daily, 2),
        "monthly_cost_usd": round(monthly, 2),
        "monthly_budget": MONTHLY_BUDGET,
        "budget_utilization_pct": round((monthly / MONTHLY_BUDGET) * 100, 1),
    }

Cost Optimization

Prompt compression

Long prompts are expensive. Every input token is paid on EVERY LLM call. If your system prompt is 3,000 tokens and you make 5 LLM calls per request, you pay 15,000 tokens for the system prompt alone.

Compress without losing intent. A 120-token system prompt that says the same thing as a 35-token one saves 71%. At 500 queries/day × 5 calls × 85 tokens saved = 212,500 tokens/day (~$0.53/day with GPT-4o). It seems small individually, but it's free optimization — the quality doesn't change.

Aggressive caching

Every cache hit is an LLM call you DON'T pay for. Cache tool results with an appropriate TTL (as you saw in capsule 03), and for exact LLM responses, use a hash of the prompt as the key:

import hashlib

async def cached_llm_call(prompt: str, model: str = "gpt-4.1-mini", ttl: int = 300) -> str | None:
    cache_key = f"llm:{model}:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
    cached = await redis_client.get(cache_key)
    return cached.decode() if cached else None

async def cache_llm_response(prompt: str, response: str, model: str = "gpt-4.1-mini", ttl: int = 300):
    cache_key = f"llm:{model}:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
    await redis_client.setex(cache_key, ttl, response)

Reducing the agent's iterations

A ReAct agent can iterate indefinitely. Limit the iterations with counters in the state: a maximum of 5 tool calls, a maximum of 2 reflection rounds, a maximum of 10 total LLM calls. When a limit is reached, jump straight to synthesis.

Optimizing the context window

Long conversations accumulate history. Use LangChain's trim_messages to keep only the recent messages within a token budget. A 4,000-token window with strategy="last" drops the oldest turns but always preserves the system prompt.

Summary of optimization techniques

Connection to the Project

The Research Agent needs complete cost control to be production-ready:

1. A token budget per request — A maximum of 50,000 tokens per research query. If it's exceeded, the agent generates a partial answer with what it has. The BudgetCallbackHandler intercepts every LLM call and aborts if it goes over.

2. Rate limiting by plan — Free: 3 req/min, 50 req/day, 500K tokens/day. Pro: 15 req/min, 1000 req/day, 5M tokens/day. Implemented with SlowAPI + Redis for the token budgets.

3. Model routing — The router (gpt-4.1-nano) classifies each query. Planning uses gpt-4.1-mini. Only synthesis and the final answer use gpt-4.1. The result: the average cost per query drops from $0.09 to $0.035.

4. Cost tracking in Redis — Every LLM call records the model, tokens, component, and user. Redis accumulates daily and monthly costs. The /costs endpoint exposes the state in real time.

5. Cost alerts — A single query > $0.50 → warning. Daily cost > $50 → warning. Daily cost > $100 → critical. Monthly budget > 75% → warning. Monthly budget > 90% → critical with a Slack notification.

6. Caching — Tool results in Redis with a 10-min TTL. Prompt compression. A 20-message window.

The result: an agent that costs ~$525/month instead of ~$1,350/month, with the same quality, and that never goes over budget.

Troubleshooting

Problem 1: "The token budget runs out before generating an answer"

Likely cause: The budget is too low, or the system prompt and tool descriptions consume too much. If your system prompt uses 2,000 tokens and you have 10 tools (200 tokens each), you've already consumed 4,000 tokens before the first real LLM call.

Solution: Audit how much each fixed component consumes. Shrink the system prompt. Load only the relevant tools (not all 10 every time). Implement dynamic budgets based on the complexity the router classified.

Problem 2: "The rate limiter blocks legitimate users"

Likely cause: You're using get_remote_address (the IP) as the key. On corporate networks, all the users share the same public IP.

Solution: Use the authenticated user_id as the key: Limiter(key_func=lambda r: r.headers.get("X-User-ID", get_remote_address(r))). Never use the IP as the sole identifier in environments with NAT/VPN.

Problem 3: "The model router picks the wrong model"

Likely cause: The router's prompt isn't clear enough, or the nano model doesn't have the capacity to classify well.

Solution: Evaluate the router with 50-100 manually classified queries. If the accuracy is < 85%, move up to mini. Add few-shot examples. Consider heuristic rules as a fallback (queries < 20 tokens → SIMPLE, queries with "analyze", "research" → COMPLEX).

Problem 4: "The reported costs don't match the OpenAI bill"

Likely cause: You're not counting all the tokens. Retries generate tokens that get charged but aren't always recorded in your callback. Embeddings for the semantic cache and health check calls add up too.

Solution: Compare your logs against OpenAI's Usage dashboard. Record tokens on retries. Include the cost of embeddings. Exclude health checks from the user metrics but include them in the total cost.

Problem 5: "Caching hurts quality on time-sensitive queries"

Likely cause: The cache's TTL is too long for queries that need real-time information.

Solution: Classify queries as "temporal" vs "atemporal" before consulting the cache. Queries with words like "today", "now", "latest", "recent" should bypass the cache or use a very short TTL (1-2 min). Add a Cache-Control: no-cache header so the user can force a fresh query.

Exercises

Exercise 1: Calculate the real cost of a multi-agent system

Your system has a supervisor that delegates to 3 sub-agents. Each sub-agent makes 2 LLM calls (planning + response) and 2 tool calls. The supervisor makes 2 LLM calls of its own (routing + merge). Use GPT-4o for everything. Calculate the cost of one query if each LLM call consumes an average of 3,000 input tokens and 800 output tokens.

Exercise 2: Design token budgets by query type

Your agent handles 3 types of queries: FAQ (a direct answer), Research (search + analysis), and Deep Analysis (multi-source + reflection). Define appropriate token budgets and what to do when each one is exceeded.

QUERY_BUDGETS = {
    "faq": {"max_tokens": 10_000, "max_llm_calls": 2, "on_exceed": "answer_from_knowledge"},
    "research": {"max_tokens": 40_000, "max_llm_calls": 6, "on_exceed": "synthesize_partial"},
    "deep_analysis": {"max_tokens": 80_000, "max_llm_calls": 12, "on_exceed": "skip_reflection"},
}

Exercise 3: Implement a cost dashboard endpoint

Implement a /costs/dashboard endpoint that returns: today's cost, this week's cost, this month's cost, the top 3 users by cost today, and the average cost per query.

from datetime import datetime, timezone, timedelta

@app.get("/costs/dashboard")
async def cost_dashboard():
    now = datetime.now(timezone.utc)
    today, month = now.strftime("%Y-%m-%d"), now.strftime("%Y-%m")

    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    week_cost = sum(
        float(await redis_client.get(f"cost:daily:{(now - timedelta(days=i)).strftime('%Y-%m-%d')}") or 0)
        for i in range(7)
    )
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)

    user_keys = await redis_client.keys(f"cost:user:*:{today}")
    user_costs = [
        {"user_id": k.decode().split(":")[2], "cost_usd": round(float(await redis_client.get(k) or 0), 4)}
        for k in user_keys
    ]
    top_users = sorted(user_costs, key=lambda x: x["cost_usd"], reverse=True)[:3]

    query_count = max(int(await redis_client.get(f"query_count:{today}") or 1), 1)
    return {
        "today_usd": round(daily, 2), "this_week_usd": round(week_cost, 2),
        "this_month_usd": round(monthly, 2), "budget_remaining_usd": round(MONTHLY_BUDGET - monthly, 2),
        "avg_cost_per_query_usd": round(daily / query_count, 4), "top_users_today": top_users,
    }

Exercise 4: Rate limiting with token awareness

Standard rate limiting counts requests, but not every request costs the same. Design a rate limiter that considers the user's accumulated cost, not just the number of requests.

async def check_cost_rate_limit(user_id: str, plan: UserPlan) -> dict:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    user_cost = float(await redis_client.get(f"cost:user:{user_id}:{today}") or 0)
    max_cost = {UserPlan.FREE: 1.00, UserPlan.PRO: 10.00, UserPlan.ENTERPRISE: 100.00}[plan]

    user_tokens = int(await redis_client.get(f"tokens:user:{user_id}:{today}") or 0)
    max_tokens = PLAN_LIMITS[plan]["daily_token_budget"]

    allowed = user_cost < max_cost and user_tokens < max_tokens
    return {
        "allowed": allowed,
        "reason": None if allowed else ("daily_cost_exceeded" if user_cost >= max_cost else "daily_tokens_exceeded"),
        "cost_today": round(user_cost, 4), "cost_limit": max_cost,
    }

Exercise 5: Compare optimization strategies

You have an agent that processes 500 queries/day at an average cost of $0.12/query ($60/day). Your goal is to get down to $30/day without noticeably degrading quality. Propose 3 strategies, estimate the savings of each, and recommend an implementation order.

Summary

• Agents are expensive by nature. Multiple LLM calls, tool calls, reflection loops, and multi-agent all multiply tokens. One query can cost $0.10-$0.50. Without control, 500 queries/day become $1,500/month.
• Token budgets are your first line of defense. A hard limit per request prevents runaway queries. The BudgetCallbackHandler intercepts every LLM call and aborts if it's exceeded.
• Rate limiting protects your budget and your users. Limit by request rate (SlowAPI) AND by daily token budget (Redis). Differentiate by plan: free, pro, enterprise.
• Model routing cuts costs 40-60% without degrading quality. Nano classifies, mini plans, full synthesizes. The savings are transformative.
• Cost monitoring with alerts prevents surprises. Track by query, user, component, model, day, month. Alert at thresholds. A /costs endpoint for real-time visibility.
• Cost optimization is a compounding effort. Prompt compression + caching + limiting iterations + a message window = a 60-75% combined reduction.
• Cost awareness is the AI Engineer's responsibility. You design the prompts, configure the models, decide the iterations. Cost is a direct consequence of your design decisions.

Next capsule: Pydantic AI — Comparison — you'll implement the same agent in Pydantic AI and compare philosophies, trade-offs, and the development experience against the LangChain/LangGraph ecosystem you've used throughout the guide.

Additional Resources

1. OpenAI Pricing — Up-to-date prices per model, input/output tokens
2. Anthropic Pricing — Claude's prices, compared with OpenAI
3. SlowAPI Documentation — Rate limiting for FastAPI based on limits
4. LangChain Callbacks — Callbacks for tracking tokens and costs
5. Redis Rate Limiting Patterns — Token buckets and sliding windows with Redis
6. Prompt Engineering for Cost Optimization — OpenAI's techniques for efficient prompts