Token Usage Tracking and Prompt Caching

Capsule overview

Tokens cost money. Every time your agent calls a model, you're paying — for every input token and every output token. In a local prototype this is irrelevant: one run of the Research Assistant costs cents. But in production, with hundreds or thousands of daily users, those cents turn into monthly bills of thousands of dollars. Without tracking, you don't know how much each user spends, which operations are the most expensive, or where to optimize.

Token tracking isn't a technical tool — it's a business tool. "This research run cost $0.12" is information that drives product decisions: can you offer this feature on the free plan? Do you need a usage limit? Is it worth optimizing the analyst prompt that eats 60% of the budget? Without these numbers, you're making decisions blind.

In the previous capsules you learned to trace with LangSmith (capsule 02), debug runs visually (capsule 03), and run automated evaluation with datasets (capsule 04). Now you add the financial dimension: how much each trace costs, each evaluation, each agent run. And with prompt caching, you can cut those costs by up to 90% on repetitive patterns.

Token tracking as a business tool

Before looking at code, understand the financial impact. These are the prices for common models (as of writing — check for updated pricing):

Now, do the math for your Research Assistant:

A typical research run:
- Decomposition (1 call): ~500 input + ~200 output = 700 tokens
- Search (4 calls):       ~2000 input + ~800 output = 2800 tokens
- Analysis (1 call):      ~3000 input + ~1000 output = 4000 tokens
- Synthesis (1 call):     ~2000 input + ~500 output = 2500 tokens
                                                    Total: ~10,000 tokens

With GPT-4.1:
- Input: 7,500 tokens × $2.00/1M = $0.015
- Output: 2,500 tokens × $8.00/1M = $0.020
- Total per research run: ~$0.035

With GPT-4.1-mini:
- Input: 7,500 tokens × $0.40/1M = $0.003
- Output: 2,500 tokens × $1.60/1M = $0.004
- Total per research run: ~$0.007

Looks like nothing. Now scale it:

1,000 users/day × 3 research runs/user = 3,000 research runs/day

With GPT-4.1:    3,000 × $0.035 = $105/day = $3,150/month
With GPT-4.1-mini: 3,000 × $0.007 = $21/day = $630/month

Is that acceptable for your business model?
How much do you charge the user?
What's your margin?

These are the questions token tracking lets you answer with data instead of intuition.

UsageMetadata: tokens per call

LangChain includes usage metadata in every model response. The usage_metadata field on the AIMessage holds the token breakdown.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke("What is LangSmith?")

print(f"Response: {response.content[:80]}...")
print(f"\nUsage metadata:")
print(f"  Input tokens:  {response.usage_metadata['input_tokens']}")
print(f"  Output tokens: {response.usage_metadata['output_tokens']}")
print(f"  Total tokens:  {response.usage_metadata['total_tokens']}")
# Expected output:
# Response: LangSmith is an observability and evaluation platform for applications ...
#
# Usage metadata:
#   Input tokens:  13
#   Output tokens: 95
#   Total tokens:  108

Every AIMessage includes usage_metadata automatically. You don't need extra configuration — just read the field.

Calculating cost per call

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

def calculate_cost(usage_metadata: dict, model_name: str) -> float:
    """Calculate the dollar cost of one model call."""
    pricing = PRICING.get(model_name, PRICING["gpt-4.1-mini"])
    input_cost = (usage_metadata["input_tokens"] / 1_000_000) * pricing["input"]
    output_cost = (usage_metadata["output_tokens"] / 1_000_000) * pricing["output"]
    return input_cost + output_cost

model = init_chat_model("openai:gpt-4.1-mini")

prompts = [
    "Say 'hello' in one word.",
    "Explain what machine learning is in 3 sentences.",
    "Write a detailed analysis of the advantages and disadvantages of microservices vs monoliths, including considerations of scalability, maintenance, testing, deployment, and operational costs.",
]

total_cost = 0.0
for prompt in prompts:
    response = model.invoke(prompt)
    cost = calculate_cost(response.usage_metadata, "gpt-4.1-mini")
    total_cost += cost
    print(f"Prompt: {prompt[:50]}...")
    print(f"  Tokens: {response.usage_metadata['input_tokens']} in / {response.usage_metadata['output_tokens']} out")
    print(f"  Cost:   ${cost:.6f}")
    print()

print(f"Total cost of 3 calls: ${total_cost:.6f}")
# Expected output:
# Prompt: Say 'hello' in one word....
#   Tokens: 14 in / 4 out
#   Cost:   $0.000012
#
# Prompt: Explain what machine learning is in 3 sentences....
#   Tokens: 16 in / 68 out
#   Cost:   $0.000115
#
# Prompt: Write a detailed analysis of the advantages and di...
#   Tokens: 40 in / 350 out
#   Cost:   $0.000576
#
# Total cost of 3 calls: $0.000703

get_openai_callback: aggregate tracking per block

To track tokens across multiple calls inside a block of code, LangChain offers get_openai_callback as a context manager.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_community.callbacks import get_openai_callback

model = init_chat_model("openai:gpt-4.1-mini")

with get_openai_callback() as cb:
    response1 = model.invoke("What is Python?")
    response2 = model.invoke("What is JavaScript?")
    response3 = model.invoke("Compare Python and JavaScript in 2 sentences.")

print(f"Total calls:   {cb.successful_requests}")
print(f"Total tokens:  {cb.total_tokens}")
print(f"  Input:       {cb.prompt_tokens}")
print(f"  Output:      {cb.completion_tokens}")
print(f"Total cost:    ${cb.total_cost:.6f}")
# Expected output:
# Total calls:   3
# Total tokens:  320
#   Input:       45
#   Output:      275
# Total cost:    $0.000458

The callback accumulates tokens and cost from every call inside the with block. It's useful for measuring the cost of a complete operation that involves multiple calls.

Tracking per operation in an agent

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_community.callbacks import get_openai_callback

model = init_chat_model("openai:gpt-4.1-mini")

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Result: {query} is a relevant topic in modern technology."

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        return str(eval(expression))
    except Exception:
        return "Error in the expression"

agent = create_agent(model, [web_search, calculate])

with get_openai_callback() as cb:
    result = agent.invoke(
        {"messages": [("user", "What is 15% of 250? Search for what you could buy with that.")]}
    )

print(f"Response: {result['messages'][-1].content[:100]}...")
print(f"\nCost of the full run:")
print(f"  Model calls:    {cb.successful_requests}")
print(f"  Total tokens:   {cb.total_tokens}")
print(f"  Cost:           ${cb.total_cost:.6f}")

COST_PER_RESEARCH = cb.total_cost
print(f"\nMonthly projection (1000 users/day, 3 runs/user):")
print(f"  Daily:   ${COST_PER_RESEARCH * 3000:.2f}")
print(f"  Monthly: ${COST_PER_RESEARCH * 3000 * 30:.2f}")
# Expected output:
# Response: 15% of 250 is 37.5. With $37.50 you could buy a range of things depending o...
#
# Cost of the full run:
#   Model calls:    3
#   Total tokens:   485
#   Cost:           $0.000680
#
# Monthly projection (1000 users/day, 3 runs/user):
#   Daily:   $2.04
#   Monthly: $61.20

UsageMetadataCallbackHandler: granular tracking

For more detailed tracking — especially when you need the breakdown per individual call inside a run — use UsageMetadataCallbackHandler.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import UsageMetadataCallbackHandler

model = init_chat_model("openai:gpt-4.1-mini")

handler = UsageMetadataCallbackHandler()

response = model.invoke(
    "Explain what observability means in AI systems.",
    config={"callbacks": [handler]},
)

print(f"Response: {response.content[:80]}...")
print(f"\nUsage accumulated in the handler:")
print(f"  Total usage records: {len(handler.usage_metadata)}")
for i, usage in enumerate(handler.usage_metadata):
    print(f"  Record {i}: {usage}")
# Expected output:
# Response: Observability in AI systems refers to the ability to understand what the ...
#
# Usage accumulated in the handler:
#   Total usage records: 1
#   Record 0: {'input_tokens': 16, 'output_tokens': 120, 'total_tokens': 136}

Tracking across multiple calls

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import UsageMetadataCallbackHandler

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}

model = init_chat_model("openai:gpt-4.1-mini")
handler = UsageMetadataCallbackHandler()
config = {"callbacks": [handler]}

operations = [
    ("decompose", "Break 'AI in education' into 3 sub-questions. Answer with the questions only."),
    ("search", "Briefly summarize what AI in education is."),
    ("analyze", "Given that AI personalizes learning and automates assessment, identify 2 key findings. Answer in 2 sentences."),
    ("write", "Write a 2-sentence executive summary about AI in education."),
]

costs_by_operation = {}
prev_count = 0

for op_name, prompt in operations:
    model.invoke(prompt, config=config)
    
    current_records = handler.usage_metadata[prev_count:]
    total_input = sum(u.get("input_tokens", 0) for u in current_records)
    total_output = sum(u.get("output_tokens", 0) for u in current_records)
    
    cost = (total_input / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (total_output / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
    
    costs_by_operation[op_name] = {
        "input_tokens": total_input,
        "output_tokens": total_output,
        "cost": cost,
    }
    prev_count = len(handler.usage_metadata)

print("Cost breakdown by operation:")
print(f"{'Operation':<12} {'Input':>8} {'Output':>8} {'Cost':>12}")
print("─" * 44)
total = 0.0
for op, data in costs_by_operation.items():
    print(f"{op:<12} {data['input_tokens']:>8} {data['output_tokens']:>8} ${data['cost']:>10.6f}")
    total += data["cost"]
print("─" * 44)
print(f"{'TOTAL':<12} {'':>8} {'':>8} ${total:>10.6f}")
# Expected output:
# Cost breakdown by operation:
# Operation      Input   Output         Cost
# ────────────────────────────────────────────
# decompose         18       45   $0.000079
# search            12       80   $0.000133
# analyze           42      60    $0.000113
# write             20       50   $0.000088
# ────────────────────────────────────────────
# TOTAL                              $0.000413

CostTracker class: production tracking

In a real system, you need a reusable tracker that accumulates costs, breaks them down by operation, and generates reports.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}


@dataclass
class CostEntry:
    operation: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


class CostTracker:
    def __init__(self, model_name: str = "gpt-4.1-mini"):
        self.model_name = model_name
        self.entries: list[CostEntry] = []

    def track(self, operation: str, usage_metadata: dict) -> CostEntry:
        """Record the cost of one operation."""
        pricing = PRICING.get(self.model_name, PRICING["gpt-4.1-mini"])
        input_tokens = usage_metadata.get("input_tokens", 0)
        output_tokens = usage_metadata.get("output_tokens", 0)

        cost = (input_tokens / 1_000_000) * pricing["input"] + \
               (output_tokens / 1_000_000) * pricing["output"]

        entry = CostEntry(
            operation=operation,
            model=self.model_name,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=cost,
        )
        self.entries.append(entry)
        return entry

    @property
    def total_cost(self) -> float:
        return sum(e.cost_usd for e in self.entries)

    @property
    def total_tokens(self) -> int:
        return sum(e.input_tokens + e.output_tokens for e in self.entries)

    def cost_by_operation(self) -> dict[str, float]:
        """Cost grouped by operation type."""
        costs: dict[str, float] = {}
        for entry in self.entries:
            costs[entry.operation] = costs.get(entry.operation, 0) + entry.cost_usd
        return costs

    def report(self) -> str:
        """Generate a human-readable cost report."""
        lines = []
        lines.append("╔══════════════════════════════════════════════╗")
        lines.append("║          COST REPORT                        ║")
        lines.append("╚══════════════════════════════════════════════╝")
        lines.append(f"  Model:          {self.model_name}")
        lines.append(f"  Operations:     {len(self.entries)}")
        lines.append(f"  Total tokens:   {self.total_tokens:,}")
        lines.append(f"  Total cost:     ${self.total_cost:.6f}")
        lines.append("")
        lines.append("  Breakdown by operation:")
        for op, cost in self.cost_by_operation().items():
            pct = (cost / self.total_cost * 100) if self.total_cost > 0 else 0
            lines.append(f"    {op:<15} ${cost:.6f}  ({pct:.1f}%)")

        daily_projection = self.total_cost * 3000
        monthly_projection = daily_projection * 30
        lines.append("")
        lines.append("  Projection (3000 runs/day):")
        lines.append(f"    Daily:   ${daily_projection:.2f}")
        lines.append(f"    Monthly: ${monthly_projection:.2f}")
        return "\n".join(lines)


tracker = CostTracker("gpt-4.1-mini")
model = init_chat_model("openai:gpt-4.1-mini")

operations = [
    ("decompose", "List 3 sub-questions about 'AI in healthcare'. Questions only."),
    ("search", "What is AI applied to medical diagnosis? Answer in 2 sentences."),
    ("search", "What advances are there in AI for drug discovery? Answer in 2 sentences."),
    ("analyze", "Given that AI improves diagnosis and accelerates drug discovery, identify 2 trends. Answer in 2 sentences."),
    ("write", "Write a 3-sentence executive summary about AI in healthcare."),
]

for op_name, prompt in operations:
    response = model.invoke(prompt)
    tracker.track(op_name, response.usage_metadata)

print(tracker.report())
# Expected output:
# ╔══════════════════════════════════════════════╗
# ║          COST REPORT                        ║
# ╚══════════════════════════════════════════════╝
#   Model:          gpt-4.1-mini
#   Operations:     5
#   Total tokens:   520
#   Total cost:     $0.000450
#
#   Breakdown by operation:
#     decompose       $0.000075  (16.7%)
#     search          $0.000180  (40.0%)
#     analyze         $0.000110  (24.4%)
#     write           $0.000085  (18.9%)
#
#   Projection (3000 runs/day):
#     Daily:   $1.35
#     Monthly: $40.50

The CostTracker answers business questions directly: "Which operation eats the most budget?" (search, 40%). "Can I offer this for free?" (at $40.50/month, maybe yes for a limited free plan). "Where do I optimize first?" (search — it's 40% of the cost).

Cost by model: choosing the right one

Not every operation needs the same model. Analysis needs reasoning — use GPT-4.1. Search and writing are simpler — use GPT-4.1-mini or nano. This pattern connects with the model routing from Module 4.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

MODEL_ASSIGNMENT = {
    "decompose": "gpt-4.1-nano",
    "search": "gpt-4.1-nano",
    "analyze": "gpt-4.1",
    "write": "gpt-4.1-mini",
}

models = {
    "gpt-4.1": init_chat_model("openai:gpt-4.1"),
    "gpt-4.1-mini": init_chat_model("openai:gpt-4.1-mini"),
    "gpt-4.1-nano": init_chat_model("openai:gpt-4.1-nano"),
}

def calculate_cost(usage: dict, model_name: str) -> float:
    pricing = PRICING[model_name]
    return (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

operations = [
    ("decompose", "List 3 sub-questions about 'AI in production'. Questions only."),
    ("search", "What is observability in AI? Answer in 1 sentence."),
    ("analyze", "Analyze why observability is critical in AI production. 2 reasons."),
    ("write", "Write 2 sentences summarizing the importance of observability in AI."),
]

total_optimized = 0.0
total_if_all_gpt4 = 0.0

print(f"{'Operation':<12} {'Model':<16} {'Tokens':>8} {'Cost':>12} {'If GPT-4.1':>12}")
print("─" * 64)

for op_name, prompt in operations:
    model_name = MODEL_ASSIGNMENT[op_name]
    model = models[model_name]

    response = model.invoke(prompt)
    usage = response.usage_metadata

    cost = calculate_cost(usage, model_name)
    cost_gpt4 = calculate_cost(usage, "gpt-4.1")

    total_optimized += cost
    total_if_all_gpt4 += cost_gpt4

    total_tokens = usage["input_tokens"] + usage["output_tokens"]
    print(f"{op_name:<12} {model_name:<16} {total_tokens:>8} ${cost:>10.6f} ${cost_gpt4:>10.6f}")

print("─" * 64)
savings = total_if_all_gpt4 - total_optimized
savings_pct = (savings / total_if_all_gpt4 * 100) if total_if_all_gpt4 > 0 else 0
print(f"{'TOTAL':<12} {'optimized':<16} {'':>8} ${total_optimized:>10.6f} ${total_if_all_gpt4:>10.6f}")
print(f"\nSavings: ${savings:.6f} ({savings_pct:.1f}%)")
print(f"Monthly projection (3000 runs/day):")
print(f"  Optimized:     ${total_optimized * 3000 * 30:.2f}/month")
print(f"  All GPT-4.1:   ${total_if_all_gpt4 * 3000 * 30:.2f}/month")
print(f"  Monthly saving: ${savings * 3000 * 30:.2f}/month")
# Expected output:
# Operation    Model              Tokens         Cost    If GPT-4.1
# ────────────────────────────────────────────────────────────────────
# decompose    gpt-4.1-nano           55   $0.000008   $0.000080
# search       gpt-4.1-nano           40   $0.000006   $0.000058
# analyze      gpt-4.1               180   $0.000456   $0.000456
# write        gpt-4.1-mini           90   $0.000060   $0.000300
# ────────────────────────────────────────────────────────────────────
# TOTAL        optimized                   $0.000530   $0.000894
#
# Savings: $0.000364 (40.7%)
# Monthly projection (3000 runs/day):
#   Optimized:     $47.70/month
#   All GPT-4.1:   $80.46/month
#   Monthly saving: $32.76/month

Prompt caching: cutting costs on repetitive patterns

Many providers offer prompt caching: if you send the same prompt prefix repeatedly, the cached tokens are billed at a reduced price (typically a 50-90% discount). This matters most for long system prompts that repeat on every call.

How caching works

Without cache:
  Call 1: [system prompt: 2000 tokens] + [user: 50 tokens] → billed 2050 input tokens
  Call 2: [system prompt: 2000 tokens] + [user: 60 tokens] → billed 2060 input tokens
  Call 3: [system prompt: 2000 tokens] + [user: 45 tokens] → billed 2045 input tokens
  Total: 6155 tokens at full price

With cache:
  Call 1: [system prompt: 2000 tokens] + [user: 50 tokens] → billed 2050 tokens (cache miss)
  Call 2: [system prompt: 2000 CACHED] + [user: 60 tokens] → billed 60 tokens + 2000 cached
  Call 3: [system prompt: 2000 CACHED] + [user: 45 tokens] → billed 45 tokens + 2000 cached
  Total: 2155 tokens at full price + 4000 tokens at reduced price

Detecting cached tokens in the response

OpenAI and Anthropic include caching information in usage_metadata when it applies. You can detect whether your prompts are benefiting from the cache.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

LONG_SYSTEM_PROMPT = """You are a senior research analyst specialized in technology.

Your analysis methodology includes:
1. Identifying primary sources and verifying their credibility
2. Detecting patterns and trends in the data
3. Evaluating contradictions between sources
4. Quantifying confidence in each finding
5. Producing actionable conclusions

Formatting rules:
- Always respond in English
- Use bullet points for findings
- Include a confidence level (high/medium/low) per finding
- Limit your response to 3-5 main findings
- Every finding must have specific evidence

Your goal is to produce analysis an executive can read in 2 minutes
and act on. Prioritize clarity over exhaustiveness."""

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "Analyze the impact of LLMs on software development.",
    "Analyze the trends in generative AI for 2026.",
    "Analyze the current state of autonomous AI agents.",
]

for i, question in enumerate(questions):
    messages = [
        SystemMessage(content=LONG_SYSTEM_PROMPT),
        HumanMessage(content=question),
    ]
    response = model.invoke(messages)
    usage = response.usage_metadata

    cached = usage.get("input_token_details", {}).get("cached", 0)
    cache_read = usage.get("cache_read_input_tokens", 0)
    cached_tokens = cached or cache_read

    print(f"Call {i+1}: {question[:50]}...")
    print(f"  Input: {usage['input_tokens']} | Output: {usage['output_tokens']} | Cached: {cached_tokens}")
# Expected output:
# Call 1: Analyze the impact of LLMs on software developmen...
#   Input: 210 | Output: 250 | Cached: 0
# Call 2: Analyze the trends in generative AI for 2026....
#   Input: 212 | Output: 230 | Cached: 192
# Call 3: Analyze the current state of autonomous AI agents...
#   Input: 214 | Output: 240 | Cached: 192

The cache kicks in automatically when the prefix matches. You don't have to do anything special with OpenAI — the system detects common prefixes and caches them.

Anthropic's explicit cache_control

With Anthropic models, you can be explicit about which parts of the prompt you want cached using cache_control.

from dotenv import load_dotenv
load_dotenv()

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage

model = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    max_tokens=500,
)

system_message = SystemMessage(
    content="""You are a senior research analyst. You always respond in English.
Limit your response to 3 main findings with a confidence level.
Every finding must have specific evidence from the provided sources.
Prioritize clarity and brevity. The executive reads your report in 2 minutes.""",
    additional_kwargs={"cache_control": {"type": "ephemeral"}},
)

questions = [
    "What are the main AI trends for 2026?",
    "What impact does AI have on the job market?",
]

for question in questions:
    response = model.invoke([system_message, HumanMessage(content=question)])
    usage = response.usage_metadata

    cache_creation = usage.get("input_token_details", {}).get("cache_creation", 0)
    cache_read = usage.get("input_token_details", {}).get("cached", 0)

    print(f"Q: {question[:50]}...")
    print(f"  Input: {usage['input_tokens']} | Cache creation: {cache_creation} | Cache read: {cache_read}")
# Expected output:
# Q: What are the main AI trends for 2026?...
#   Input: 95 | Cache creation: 80 | Cache read: 0
# Q: What impact does AI have on the job market?...
#   Input: 95 | Cache creation: 0 | Cache read: 80

Token budgets: limits per user

In production, you need to set token limits per user to prevent runaway costs. A user who runs 100 research queries in an hour can drain your daily budget.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime, timedelta

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}


@dataclass
class UserBudget:
    user_id: str
    daily_limit_usd: float
    monthly_limit_usd: float
    spent_today_usd: float = 0.0
    spent_this_month_usd: float = 0.0
    last_reset_daily: str = field(default_factory=lambda: datetime.now().date().isoformat())
    last_reset_monthly: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m"))

    def can_spend(self, estimated_cost: float) -> tuple[bool, str]:
        """Check whether the user can afford the estimated cost."""
        if self.spent_today_usd + estimated_cost > self.daily_limit_usd:
            return False, f"Daily limit reached (${self.spent_today_usd:.4f}/${self.daily_limit_usd:.4f})"
        if self.spent_this_month_usd + estimated_cost > self.monthly_limit_usd:
            return False, f"Monthly limit reached (${self.spent_this_month_usd:.2f}/${self.monthly_limit_usd:.2f})"
        return True, "OK"

    def record_spend(self, cost: float):
        """Record a charge."""
        self.spent_today_usd += cost
        self.spent_this_month_usd += cost

    def alert_level(self) -> str:
        """Return the alert level based on spend."""
        daily_pct = (self.spent_today_usd / self.daily_limit_usd * 100) if self.daily_limit_usd > 0 else 0
        if daily_pct >= 100:
            return "BLOCKED"
        if daily_pct >= 80:
            return "WARNING"
        if daily_pct >= 50:
            return "NOTICE"
        return "OK"


TIER_LIMITS = {
    "free": {"daily": 0.05, "monthly": 1.00},
    "pro": {"daily": 0.50, "monthly": 10.00},
    "enterprise": {"daily": 5.00, "monthly": 100.00},
}


def create_user_budget(user_id: str, tier: str) -> UserBudget:
    limits = TIER_LIMITS[tier]
    return UserBudget(
        user_id=user_id,
        daily_limit_usd=limits["daily"],
        monthly_limit_usd=limits["monthly"],
    )


model = init_chat_model("openai:gpt-4.1-mini")

free_user = create_user_budget("user-free-001", "free")
pro_user = create_user_budget("user-pro-001", "pro")

users = [free_user, pro_user]

for user in users:
    print(f"\n{'='*50}")
    print(f"User: {user.user_id} (daily limit: ${user.daily_limit_usd})")
    print(f"{'='*50}")

    for i in range(5):
        estimated_cost = 0.01
        can_spend, reason = user.can_spend(estimated_cost)

        if not can_spend:
            print(f"  Attempt {i+1}: BLOCKED — {reason}")
            continue

        response = model.invoke(f"Question {i+1}: What is observability?")
        usage = response.usage_metadata
        actual_cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
                      (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]

        user.record_spend(actual_cost)
        alert = user.alert_level()
        print(f"  Attempt {i+1}: ${actual_cost:.6f} | Total: ${user.spent_today_usd:.6f} | Alert: {alert}")
# Expected output:
# ==================================================
# User: user-free-001 (daily limit: $0.05)
# ==================================================
#   Attempt 1: $0.000150 | Total: $0.000150 | Alert: OK
#   Attempt 2: $0.000145 | Total: $0.000295 | Alert: OK
#   Attempt 3: $0.000148 | Total: $0.000443 | Alert: OK
#   Attempt 4: $0.000152 | Total: $0.000595 | Alert: OK
#   Attempt 5: $0.000149 | Total: $0.000744 | Alert: OK
#
# ==================================================
# User: user-pro-001 (daily limit: $0.5)
# ==================================================
#   Attempt 1: $0.000150 | Total: $0.000150 | Alert: OK
#   Attempt 2: $0.000145 | Total: $0.000295 | Alert: OK
#   Attempt 3: $0.000148 | Total: $0.000443 | Alert: OK
#   Attempt 4: $0.000152 | Total: $0.000595 | Alert: OK
#   Attempt 5: $0.000149 | Total: $0.000744 | Alert: OK

Cost optimization strategies

Cutting costs isn't only about picking a cheaper model. There are several levers you can combine.

Strategy 1: Shorter prompts

Every token in the prompt costs money. A 500-token system prompt vs a 200-token one is a 2.5x difference in input cost — on every single call.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

verbose_prompt = """You are a highly qualified and experienced research assistant.
Your role is to help users find relevant and accurate information about
any topic they ask you about. You must be detailed in your answers but also
concise. Always verify the facts before answering. If you're unsure about something,
state it clearly. Always respond in English and use a professional but
approachable tone. Structure your answers with bullet points when appropriate."""

concise_prompt = """Research assistant. Respond in English, concise, with bullet points. Flag uncertainty if it applies."""

question = "What are the advantages of using LangGraph?"

response_verbose = model.invoke([
    SystemMessage(content=verbose_prompt),
    HumanMessage(content=question),
])

response_concise = model.invoke([
    SystemMessage(content=concise_prompt),
    HumanMessage(content=question),
])

verbose_input = response_verbose.usage_metadata["input_tokens"]
concise_input = response_concise.usage_metadata["input_tokens"]
saved = verbose_input - concise_input

print(f"Verbose system prompt: {verbose_input} input tokens")
print(f"Concise system prompt: {concise_input} input tokens")
print(f"Tokens saved:          {saved} ({saved/verbose_input*100:.0f}%)")
print(f"\nAt 3000 calls/day × 30 days × $0.40/1M tokens:")
print(f"  Verbose: ${verbose_input * 90000 / 1_000_000 * 0.40:.2f}/month")
print(f"  Concise: ${concise_input * 90000 / 1_000_000 * 0.40:.2f}/month")
# Expected output:
# Verbose system prompt: 98 input tokens
# Concise system prompt: 30 input tokens
# Tokens saved:          68 (69%)
#
# At 3000 calls/day × 30 days × $0.40/1M tokens:
#   Verbose: $3.53/month
#   Concise: $1.08/month

Strategy 2: Message trimming (connects with M8)

When you use memory, the message history grows with every turn. Without trimming, after 20 turns you can end up with 5000+ tokens of history sent on every call.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage, trim_messages,
)

messages = [
    SystemMessage(content="You are a research assistant."),
    HumanMessage(content="What is Python?"),
    AIMessage(content="Python is a high-level programming language..."),
    HumanMessage(content="And JavaScript?"),
    AIMessage(content="JavaScript is a programming language that runs in the browser..."),
    HumanMessage(content="Which is better for backend?"),
    AIMessage(content="For backend, Python has frameworks like Django and FastAPI..."),
    HumanMessage(content="And for AI?"),
    AIMessage(content="For AI and machine learning, Python dominates with libraries like..."),
    HumanMessage(content="Summarize everything above in 2 sentences."),
]

print(f"Total messages before trim: {len(messages)}")

trimmed = trim_messages(
    messages,
    max_tokens=200,
    strategy="last",
    token_counter=len,
    include_system=True,
    start_on="human",
)

print(f"Total messages after trim: {len(trimmed)}")
print(f"\nMessages kept:")
for msg in trimmed:
    role = msg.__class__.__name__.replace("Message", "")
    print(f"  [{role}] {msg.content[:60]}...")
# Expected output:
# Total messages before trim: 10
# Total messages after trim: 3
#
# Messages kept:
#   [System] You are a research assistant....
#   [AI] For AI and machine learning, Python dominates with libraries like...
#   [Human] Summarize everything above in 2 sentences....

Strategy 3: Cheap model for simple tasks (connects with M4)

You already saw this in detail in Module 4 (Dynamic Models). The principle: use the cheapest model that can do the job well.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

scenarios = {
    "All GPT-4.1":       {"decompose": "gpt-4.1", "search": "gpt-4.1", "analyze": "gpt-4.1", "write": "gpt-4.1"},
    "Optimized mix":     {"decompose": "gpt-4.1-nano", "search": "gpt-4.1-nano", "analyze": "gpt-4.1", "write": "gpt-4.1-mini"},
    "All GPT-4.1-nano":  {"decompose": "gpt-4.1-nano", "search": "gpt-4.1-nano", "analyze": "gpt-4.1-nano", "write": "gpt-4.1-nano"},
}

avg_tokens_per_op = {
    "decompose": {"input": 200, "output": 100},
    "search": {"input": 150, "output": 200},
    "analyze": {"input": 1500, "output": 500},
    "write": {"input": 800, "output": 400},
}

print(f"{'Scenario':<22} {'Cost/run':>16} {'Monthly (90K)':>14} {'Quality':>10}")
print("─" * 66)

for scenario_name, assignment in scenarios.items():
    total = 0.0
    for op, model_name in assignment.items():
        pricing = PRICING[model_name]
        tokens = avg_tokens_per_op[op]
        cost = (tokens["input"] / 1_000_000) * pricing["input"] + \
               (tokens["output"] / 1_000_000) * pricing["output"]
        total += cost

    monthly = total * 90000
    quality = "High" if "gpt-4.1" in assignment.get("analyze", "") and assignment["analyze"] == "gpt-4.1" else "Medium" if "mini" in assignment.get("analyze", "") else "Low"
    print(f"{scenario_name:<22} ${total:>14.6f} ${monthly:>12.2f} {quality:>10}")
# Expected output:
# Scenario                      Cost/run  Monthly (90K)    Quality
# ──────────────────────────────────────────────────────────────────
# All GPT-4.1            $0.011700       $1053.00       High
# Optimized mix          $0.004035        $363.15       High
# All GPT-4.1-nano       $0.000530         $47.70        Low

The "Optimized mix" row is the sweet spot: high quality where it matters (analysis), low cost where it doesn't affect quality (decomposition, search, simple writing).

Cost monitoring: dashboards and alerts

In production, you need continuous visibility into spend. LangSmith gives you cost dashboards out of the box once tracing is enabled. But you can also build your own monitoring.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class DailyUsageRecord:
    date: str
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    num_requests: int = 0
    by_model: dict = field(default_factory=dict)

class UsageMonitor:
    def __init__(self, daily_budget: float, alert_threshold: float = 0.8):
        self.daily_budget = daily_budget
        self.alert_threshold = alert_threshold
        self.records: dict[str, DailyUsageRecord] = {}

    def _today_key(self) -> str:
        return datetime.now().date().isoformat()

    def _get_today(self) -> DailyUsageRecord:
        key = self._today_key()
        if key not in self.records:
            self.records[key] = DailyUsageRecord(date=key)
        return self.records[key]

    def record(self, model: str, tokens: int, cost: float):
        today = self._get_today()
        today.total_tokens += tokens
        today.total_cost_usd += cost
        today.num_requests += 1
        today.by_model[model] = today.by_model.get(model, 0.0) + cost

        usage_pct = today.total_cost_usd / self.daily_budget
        if usage_pct >= 1.0:
            print(f"  [ALERT] BUDGET EXCEEDED: ${today.total_cost_usd:.4f}/${self.daily_budget:.4f}")
        elif usage_pct >= self.alert_threshold:
            print(f"  [WARNING] Budget at {usage_pct:.0%}: ${today.total_cost_usd:.4f}/${self.daily_budget:.4f}")

    def daily_report(self) -> str:
        today = self._get_today()
        pct = (today.total_cost_usd / self.daily_budget * 100) if self.daily_budget > 0 else 0
        lines = [
            f"Daily Usage Report — {today.date}",
            f"  Requests: {today.num_requests}",
            f"  Tokens:   {today.total_tokens:,}",
            f"  Cost:     ${today.total_cost_usd:.4f} / ${self.daily_budget:.4f} ({pct:.1f}%)",
            f"  By model:",
        ]
        for model, cost in today.by_model.items():
            lines.append(f"    {model}: ${cost:.4f}")
        return "\n".join(lines)


monitor = UsageMonitor(daily_budget=0.001, alert_threshold=0.8)

simulated_calls = [
    ("gpt-4.1-mini", 150, 0.00015),
    ("gpt-4.1-mini", 200, 0.00020),
    ("gpt-4.1", 500, 0.00080),
    ("gpt-4.1-mini", 180, 0.00018),
]

for model, tokens, cost in simulated_calls:
    print(f"Recording: {model} | {tokens} tokens | ${cost:.5f}")
    monitor.record(model, tokens, cost)

print(f"\n{monitor.daily_report()}")
# Expected output:
# Recording: gpt-4.1-mini | 150 tokens | $0.00015
# Recording: gpt-4.1-mini | 200 tokens | $0.00020
# Recording: gpt-4.1 | 500 tokens | $0.00080
#   [WARNING] Budget at 115%: $0.00115/$0.00100
# Recording: gpt-4.1-mini | 180 tokens | $0.00018
#   [ALERT] BUDGET EXCEEDED: $0.00133/$0.00100
#
# Daily Usage Report — 2026-03-08
#   Requests: 4
#   Tokens:   1,030
#   Cost:     $0.0013 / $0.0010 (133.0%)
#   By model:
#     gpt-4.1-mini: $0.0005
#     gpt-4.1: $0.0008

Troubleshooting

Problem 1: usage_metadata returns None or is empty

Cause: Not every provider includes usage metadata by default. Some require explicit configuration.

Fix: Check that the provider supports usage metadata. For OpenAI and Anthropic it works by default. For other providers, check the documentation:

response = model.invoke("test")
if response.usage_metadata:
    print(f"Tokens: {response.usage_metadata}")
else:
    print("Usage metadata not available for this provider")

Problem 2: The costs you calculate don't match the provider's invoice

Cause: Prices change frequently, or you're using stale prices in your PRICING table.

Fix: Always check current prices on the provider's page. Consider keeping prices in a config file that gets updated periodically:

PRICING_LAST_UPDATED = "2026-03-01"

Problem 3: The get_openai_callback callback doesn't capture Anthropic calls

Cause: get_openai_callback is specific to OpenAI models. It doesn't work with other providers.

Fix: Use UsageMetadataCallbackHandler, which is provider-agnostic, or read response.usage_metadata directly on each call.

Problem 4: Cached tokens don't show up in the breakdown

Cause: Prompt caching requires the prefix to hit a minimum length (typically 1024+ tokens for OpenAI). Short prompts never get cached.

Fix: Automatic caching only kicks in with prompts that are long enough. If your system prompt is under 1024 tokens, you won't see any caching benefit. For Anthropic, use explicit cache_control.

Problem 5: The CostTracker doesn't survive restarts

Cause: The in-memory implementation is lost when the process ends.

Fix: For production, persist the data. LangSmith already does this for you if tracing is enabled. If you need custom tracking, save it to a database:

import json

def save_tracker(tracker, filepath="cost_log.json"):
    data = [{"op": e.operation, "cost": e.cost_usd, "ts": e.timestamp} for e in tracker.entries]
    with open(filepath, "w") as f:
        json.dump(data, f)

Exercises

Exercise 1: Calculate the cost of a conversation (Easy)

Write a function that takes a list of model responses and computes the total cost. Test it with 3 model calls using prompts of different lengths.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

def total_cost_from_responses(responses: list, model_name: str) -> dict:
    """Compute the total cost of a list of responses."""
    total_input = 0
    total_output = 0
    pricing = PRICING[model_name]

    for r in responses:
        total_input += r.usage_metadata["input_tokens"]
        total_output += r.usage_metadata["output_tokens"]

    input_cost = (total_input / 1_000_000) * pricing["input"]
    output_cost = (total_output / 1_000_000) * pricing["output"]

    return {
        "total_input_tokens": total_input,
        "total_output_tokens": total_output,
        "input_cost": input_cost,
        "output_cost": output_cost,
        "total_cost": input_cost + output_cost,
    }

model = init_chat_model("openai:gpt-4.1-mini")

responses = [
    model.invoke("Hello"),
    model.invoke("What is Python? Answer in 1 sentence."),
    model.invoke("Explain what machine learning is, its main types, and give an example of each."),
]

result = total_cost_from_responses(responses, "gpt-4.1-mini")
print(f"Input tokens:  {result['total_input_tokens']}")
print(f"Output tokens: {result['total_output_tokens']}")
print(f"Input cost:    ${result['input_cost']:.6f}")
print(f"Output cost:   ${result['output_cost']:.6f}")
print(f"Total cost:    ${result['total_cost']:.6f}")
# Expected output:
# Input tokens:  50
# Output tokens: 200
# Input cost:    $0.000020
# Output cost:   $0.000320
# Total cost:    $0.000340

Exercise 2: CostTracker with a daily budget (Easy)

Extend CostTracker so it accepts a daily budget and raises an alert when it hits 80%.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

class BudgetCostTracker:
    def __init__(self, model_name: str, daily_budget: float):
        self.model_name = model_name
        self.daily_budget = daily_budget
        self.total_cost = 0.0
        self.call_count = 0

    def track(self, usage_metadata: dict) -> dict:
        pricing = PRICING[self.model_name]
        cost = (usage_metadata["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage_metadata["output_tokens"] / 1_000_000) * pricing["output"]
        self.total_cost += cost
        self.call_count += 1

        pct = (self.total_cost / self.daily_budget * 100) if self.daily_budget > 0 else 0
        alert = None
        if pct >= 100:
            alert = "BLOCKED"
        elif pct >= 80:
            alert = "WARNING"

        return {"cost": cost, "total": self.total_cost, "pct": pct, "alert": alert}

tracker = BudgetCostTracker("gpt-4.1-mini", daily_budget=0.001)
model = init_chat_model("openai:gpt-4.1-mini")

for i in range(8):
    response = model.invoke(f"Question {i+1}: Explain an AI concept in 2 sentences.")
    result = tracker.track(response.usage_metadata)
    alert_str = f" [{result['alert']}]" if result['alert'] else ""
    print(f"Call {i+1}: ${result['cost']:.6f} | Total: ${result['total']:.6f} ({result['pct']:.1f}%){alert_str}")
# Expected output:
# Call 1: $0.000120 | Total: $0.000120 (12.0%)
# Call 2: $0.000115 | Total: $0.000235 (23.5%)
# Call 3: $0.000118 | Total: $0.000353 (35.3%)
# Call 4: $0.000122 | Total: $0.000475 (47.5%)
# Call 5: $0.000119 | Total: $0.000594 (59.4%)
# Call 6: $0.000125 | Total: $0.000719 (71.9%)
# Call 7: $0.000121 | Total: $0.000840 (84.0%) [WARNING]
# Call 8: $0.000118 | Total: $0.000958 (95.8%) [WARNING]

Exercise 3: Compare the cost of 3 models on the same task (Medium)

Send the same prompt to GPT-4.1, GPT-4.1-mini, and GPT-4.1-nano. Compare tokens used, cost, and response quality.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

model_names = ["gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"]
prompt = "Explain the 3 main advantages of using LangGraph to build AI agents. Be concise."

print(f"Prompt: {prompt}\n")
print(f"{'Model':<16} {'Input':>7} {'Output':>7} {'Cost':>12} {'Response (first 80 chars)'}")
print("─" * 90)

for model_name in model_names:
    model = init_chat_model(f"openai:{model_name}")
    response = model.invoke(prompt)
    usage = response.usage_metadata
    pricing = PRICING[model_name]

    cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

    preview = response.content[:80].replace("\n", " ")
    print(f"{model_name:<16} {usage['input_tokens']:>7} {usage['output_tokens']:>7} ${cost:>10.6f} {preview}...")
# Expected output:
# Prompt: Explain the 3 main advantages of using LangGraph to build AI agents. Be concise.
#
# Model              Input  Output         Cost Response (first 80 chars)
# ──────────────────────────────────────────────────────────────────────────────────────────
# gpt-4.1               22     180   $0.001484 1. **Granular flow control**: LangGraph lets you define state graphs tha...
# gpt-4.1-mini          22     150   $0.000249 1. **Precise flow control**: Lets you design the agent's flow as a graph...
# gpt-4.1-nano          22     120   $0.000050 1. Flow control: Define graphs to handle the agent's logic in a structur...

Exercise 4: A tracker showing the percentage breakdown per operation (Medium)

Build a tracker that runs the Research Assistant's 4 operations (decompose, search, analyze, write) and shows what percentage of the total cost each one represents.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

class OperationTracker:
    def __init__(self):
        self.operations: dict[str, float] = {}

    def track(self, operation: str, usage: dict):
        cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
               (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
        self.operations[operation] = self.operations.get(operation, 0.0) + cost

    def breakdown(self):
        total = sum(self.operations.values())
        print(f"\nCost Breakdown (Total: ${total:.6f})")
        print(f"{'Operation':<15} {'Cost':>12} {'Share':>8} {'Bar'}")
        print("─" * 50)
        for op, cost in sorted(self.operations.items(), key=lambda x: -x[1]):
            pct = (cost / total * 100) if total > 0 else 0
            bar = "█" * int(pct / 2)
            print(f"{op:<15} ${cost:>10.6f} {pct:>6.1f}% {bar}")

tracker = OperationTracker()
model = init_chat_model("openai:gpt-4.1-mini")

ops = [
    ("decompose", "Generate 3 sub-questions about 'AI in medicine'. Questions only, one per line."),
    ("search", "What is AI applied to diagnosis? 1 sentence."),
    ("search", "What advances are there in AI for drug discovery? 1 sentence."),
    ("search", "How is AI used in medical imaging? 1 sentence."),
    ("analyze", "Given that AI improves diagnosis, accelerates drug discovery, and analyzes medical images, identify 3 main trends. Explain each in 2 sentences."),
    ("write", "Write a 4-sentence executive summary about AI in medicine, covering diagnosis, pharma, and medical imaging."),
]

for op_name, prompt in ops:
    response = model.invoke(prompt)
    tracker.track(op_name, response.usage_metadata)

tracker.breakdown()
# Expected output:
# Cost Breakdown (Total: $0.000680)
# Operation             Cost    Share Bar
# ──────────────────────────────────────────────────
# analyze         $0.000250  36.8% ██████████████████
# write           $0.000165  24.3% ████████████
# search          $0.000180  26.5% █████████████
# decompose       $0.000085  12.5% ██████

Exercise 5: A token budget system by user tier (Hard)

Implement a system where "free" users get a $0.01/day budget, "pro" users $0.10/day, and "enterprise" users $1.00/day. Simulate 10 requests per user and show who gets blocked first.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}
TIER_BUDGETS = {"free": 0.001, "pro": 0.010, "enterprise": 0.100}

class TieredUser:
    def __init__(self, user_id: str, tier: str):
        self.user_id = user_id
        self.tier = tier
        self.budget = TIER_BUDGETS[tier]
        self.spent = 0.0
        self.successful_requests = 0
        self.blocked_requests = 0

    def try_request(self, cost: float) -> bool:
        if self.spent + cost > self.budget:
            self.blocked_requests += 1
            return False
        self.spent += cost
        self.successful_requests += 1
        return True

    def status(self) -> str:
        pct = (self.spent / self.budget * 100) if self.budget > 0 else 0
        return (f"{self.user_id} ({self.tier}): "
                f"${self.spent:.6f}/${self.budget:.4f} ({pct:.0f}%) | "
                f"OK: {self.successful_requests} | Blocked: {self.blocked_requests}")

model = init_chat_model("openai:gpt-4.1-mini")
users = [
    TieredUser("alice", "free"),
    TieredUser("bob", "pro"),
    TieredUser("corp-1", "enterprise"),
]

prompts = [f"Explain AI engineering concept #{i+1} in 2 sentences." for i in range(10)]

for prompt in prompts:
    response = model.invoke(prompt)
    usage = response.usage_metadata
    cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]

    for user in users:
        user.try_request(cost)

print("Final Status:")
print("─" * 70)
for user in users:
    print(user.status())
# Expected output:
# Final Status:
# ──────────────────────────────────────────────────────────────────────
# alice (free): $0.000950/0.0010 (95%) | OK: 8 | Blocked: 2
# bob (pro): $0.001200/0.0100 (12%) | OK: 10 | Blocked: 0
# corp-1 (enterprise): $0.001200/0.1000 (1%) | OK: 10 | Blocked: 0

Exercise 6: A full dashboard with monthly projection (Hard)

Build a dashboard that combines per-operation tracking, per-model tracking, and generates a monthly cost projection based on current usage. Simulate a complete Research Assistant flow.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}

class ProductionDashboard:
    def __init__(self):
        self.entries = []

    def track(self, operation: str, model_name: str, usage: dict):
        pricing = PRICING.get(model_name, PRICING["gpt-4.1-mini"])
        cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage["output_tokens"] / 1_000_000) * pricing["output"]
        self.entries.append({
            "operation": operation,
            "model": model_name,
            "input_tokens": usage["input_tokens"],
            "output_tokens": usage["output_tokens"],
            "cost": cost,
            "timestamp": datetime.now().isoformat(),
        })

    def render(self, executions_per_day: int = 3000):
        total_cost = sum(e["cost"] for e in self.entries)
        total_tokens = sum(e["input_tokens"] + e["output_tokens"] for e in self.entries)

        by_op = {}
        for e in self.entries:
            by_op[e["operation"]] = by_op.get(e["operation"], 0.0) + e["cost"]

        by_model = {}
        for e in self.entries:
            by_model[e["model"]] = by_model.get(e["model"], 0.0) + e["cost"]

        print(f"╔{'═'*56}╗")
        print(f"║{'PRODUCTION COST DASHBOARD':^56}║")
        print(f"╚{'═'*56}╝")
        print(f"  This execution:")
        print(f"    Calls:  {len(self.entries)}")
        print(f"    Tokens: {total_tokens:,}")
        print(f"    Cost:   ${total_cost:.6f}")
        print(f"\n  By operation:")
        for op, cost in sorted(by_op.items(), key=lambda x: -x[1]):
            pct = cost / total_cost * 100 if total_cost > 0 else 0
            print(f"    {op:<15} ${cost:.6f} ({pct:.0f}%)")
        print(f"\n  By model:")
        for model, cost in sorted(by_model.items(), key=lambda x: -x[1]):
            pct = cost / total_cost * 100 if total_cost > 0 else 0
            print(f"    {model:<16} ${cost:.6f} ({pct:.0f}%)")
        daily = total_cost * executions_per_day
        monthly = daily * 30
        print(f"\n  Projection ({executions_per_day:,} exec/day):")
        print(f"    Daily:   ${daily:.2f}")
        print(f"    Monthly: ${monthly:.2f}")
        print(f"    Yearly:  ${monthly * 12:.2f}")

dashboard = ProductionDashboard()

mini = init_chat_model("openai:gpt-4.1-mini")
strong = init_chat_model("openai:gpt-4.1")

ops = [
    ("decompose", mini, "gpt-4.1-mini", "List 3 sub-questions about 'AI in production'. Questions only."),
    ("search", mini, "gpt-4.1-mini", "What is observability in AI? 1 sentence."),
    ("search", mini, "gpt-4.1-mini", "What is rate limiting in APIs? 1 sentence."),
    ("analyze", strong, "gpt-4.1", "Analyze how observability and rate limiting combine to make AI systems production-ready. 3 sentences."),
    ("write", mini, "gpt-4.1-mini", "Write a 3-sentence summary about getting AI systems ready for production."),
]

for op_name, model, model_name, prompt in ops:
    response = model.invoke(prompt)
    dashboard.track(op_name, model_name, response.usage_metadata)

dashboard.render()
# Expected output:
# ╔════════════════════════════════════════════════════════╗
# ║              PRODUCTION COST DASHBOARD                ║
# ╚════════════════════════════════════════════════════════╝
#   This execution:
#     Calls:  5
#     Tokens: 480
#     Cost:   $0.000850
#
#   By operation:
#     analyze         $0.000520 (61%)
#     search          $0.000140 (16%)
#     write           $0.000110 (13%)
#     decompose       $0.000080 (9%)
#
#   By model:
#     gpt-4.1          $0.000520 (61%)
#     gpt-4.1-mini     $0.000330 (39%)
#
#   Projection (3,000 exec/day):
#     Daily:   $2.55
#     Monthly: $76.50
#     Yearly:  $918.00

Summary

In this capsule you learned:

• Token tracking is a business tool — "this research run cost $0.035" isn't a technical number, it's the number that determines pricing, margins, and product viability
• usage_metadata on every AIMessage gives you the token breakdown per individual call
• get_openai_callback accumulates tokens and cost across every call inside a with block
• UsageMetadataCallbackHandler offers granular tracking that works across multiple providers
• CostTracker as a reusable class gives you per-operation breakdowns, monthly projections, and business reports
• Model routing by cost (GPT-4.1 for analysis, nano for simple tasks) can cut costs 40-65% without sacrificing quality where it matters
• Prompt caching cuts the cost of repeated input tokens by up to 90% — especially valuable with long system prompts
• Per-user token budgets prevent runaway costs and enable tiered business models (free/pro/enterprise)
• Message trimming keeps the context lean and reduces accumulated costs in long conversations

Next capsule: Rate Limiting and Cost Control — how to keep a single user from draining your daily budget.

Additional resources

1. LangSmith Cost Tracking — Cost dashboard integrated with tracing
2. OpenAI Token Usage — OpenAI usage metadata documentation
3. Anthropic Prompt Caching — Official prompt caching guide with Claude
4. OpenAI Pricing — Up-to-date model pricing
5. LangChain Callbacks Guide — The callback system for tracking and monitoring
6. LangChain Message Trimming — How to trim messages to optimize tokens

Module 12 — LangChain & LangGraph: From Chains to Agents