Performance Optimization

By the end of this lesson, you'll apply token optimization strategies, implement caching patterns for tool outputs, track performance metrics that reveal cost and latency bottlenecks, and attribute spend to business units with granular cost tracking.

Why Performance Optimization Matters

Every token costs money, every millisecond costs patience. Research from early adopters of agentic systems reports a 20% reduction in development time and 15-30% improvement in token efficiency when teams apply structured optimization. The difference between a prototype agent and a production agent is often not capability — it's cost control and latency management.

The DAIR.AI Prompt Engineering Guide identifies token efficiency as a first-class concern: prompts that are verbose, redundant, or poorly structured waste budget on every invocation without improving output quality. Optimization is not premature here — it's table stakes for production deployment.

Token Optimization Strategies

Tokens are your primary cost driver. Four strategies reduce consumption without degrading results.

1. Concise Schemas

Replace verbose natural-language instructions with structured schemas. The model parses compact formats more efficiently than prose descriptions of the same constraints:

# ❌ Verbose — 147 tokens
"""
Please return your response as a JSON object. The object should have
a field called "analysis" which contains your analysis as a string.
It should also have a field called "confidence" which is a number
between 0 and 1 representing how confident you are. Finally, include
a field called "sources" which is a list of strings, where each
string is a URL you referenced.
"""

# ✅ Concise schema — 42 tokens
"""
Respond in JSON:
{"analysis": str, "confidence": float 0-1, "sources": [str]}
"""

The concise version saves 105 tokens per request. At 10,000 requests per day, that's over a million tokens saved daily — purely from tighter formatting.

2. Prompt Caching

Most API providers now support prompt caching: identical prompt prefixes are stored server-side and not re-processed on subsequent calls. Structure your prompts with a stable prefix (system instructions, tool definitions) and a variable suffix (user input, context):

# Structure prompts for cache hits
# The stable prefix is cached after the first request

SYSTEM_PREFIX = """You are a code review agent. You analyze diffs for:
- Security vulnerabilities
- Performance regressions
- Style violations
- Missing test coverage

Output format: {"issues": [{"severity": "high|medium|low", "line": int, "msg": str}]}
"""

def build_prompt(diff: str) -> list[dict]:
    return [
        {"role": "system", "content": SYSTEM_PREFIX},  # cached after first call
        {"role": "user", "content": f"Review this diff:\n{diff}"},  # variable per request
    ]

Prompt caching typically reduces latency by 50-80% for the cached portion and costs significantly less per cached token. The key constraint: the prefix must be identical byte-for-byte. Any variation — even a timestamp or request ID embedded in the system prompt — breaks the cache.

3. Context Compression via Summarization

Long conversation histories and large documents consume tokens that add diminishing value. Compress older context into summaries while preserving recent detail:

from dataclasses import dataclass

@dataclass
class ConversationManager:
    """Manages context window by compressing old messages."""
    max_tokens: int = 100_000
    summary_threshold: int = 80_000  # compress when 80% full
    recent_window: int = 10          # keep last N messages verbatim

    def compress_context(self, messages: list[dict]) -> list[dict]:
        total_tokens = sum(count_tokens(m["content"]) for m in messages)

        if total_tokens < self.summary_threshold:
            return messages  # no compression needed

        # Keep system prompt and recent messages verbatim
        system = [m for m in messages if m["role"] == "system"]
        recent = messages[-self.recent_window:]
        old = messages[len(system):-self.recent_window]

        # Summarize old messages into a single context block
        summary = self._summarize(old)
        compressed = system + [
            {"role": "system", "content": f"Previous context summary:\n{summary}"}
        ] + recent

        return compressed

    def _summarize(self, messages: list[dict]) -> str:
        """Distill key decisions, facts, and pending items."""
        content = "\n".join(m["content"] for m in messages)
        # Use a fast, cheap model for summarization
        return call_model(
            model="fast-summarizer",
            prompt=f"Summarize the key facts, decisions, and open items:\n{content}",
            max_tokens=500,
        )

This pattern trades a small summarization cost for large savings on subsequent calls. A 50-message history compressed to a 500-token summary saves tens of thousands of tokens on every subsequent request that includes that context.

4. Remove Redundant Instructions

Audit your prompts for instructions that repeat what the model already knows, restate constraints already encoded in the schema, or duplicate information present in tool definitions:

# ❌ Redundant — repeats what the schema and tools already specify
"""
You have access to a search tool. When you want to search, use the
search tool. The search tool takes a query parameter which should be
a string. Make sure to provide a query when using the search tool.
After searching, analyze the results. The results will be returned
as a list. Process each item in the list.
Remember to always provide your response in JSON format as specified
in the output schema above.
"""

# ✅ Lean — trusts the model to read tool definitions and schemas
"""
Search for relevant information, then analyze the results.
"""

The DAIR.AI guide emphasizes that modern models read tool schemas and output format specifications directly. Restating them in prose wastes tokens and can even confuse the model when the prose diverges from the actual schema.

Caching Patterns for Tool Outputs

Tool calls are expensive — they incur network latency, API rate limits, and sometimes monetary cost. Cache results for operations that produce stable outputs:

import time
from functools import lru_cache
from hashlib import sha256

class ToolCache:
    """Cache tool outputs to avoid redundant API calls."""

    def __init__(self, ttl_seconds: int = 300):
        self.ttl = ttl_seconds
        self._cache: dict[str, tuple[float, any]] = {}

    def _key(self, tool_name: str, params: dict) -> str:
        raw = f"{tool_name}:{sorted(params.items())}"
        return sha256(raw.encode()).hexdigest()

    def get(self, tool_name: str, params: dict):
        key = self._key(tool_name, params)
        if key in self._cache:
            timestamp, value = self._cache[key]
            if time.time() - timestamp < self.ttl:
                return value
            del self._cache[key]
        return None

    def set(self, tool_name: str, params: dict, value):
        key = self._key(tool_name, params)
        self._cache[key] = (time.time(), value)


# Simple lru_cache for pure tool outputs
@lru_cache(maxsize=256)
def fetch_file_content(repo: str, path: str, commit_sha: str) -> str:
    """Fetch file content — immutable for a given commit SHA."""
    return api_client.get_file(repo, path, commit_sha)

@lru_cache(maxsize=128)
def resolve_package_version(package: str, version_set: str) -> str:
    """Resolve package version — stable within a version set."""
    return brazil_client.resolve(package, version_set)


# Usage in agent tool execution
tool_cache = ToolCache(ttl_seconds=600)

def execute_tool(tool_name: str, params: dict) -> dict:
    """Execute a tool with caching for stable operations."""
    # Check cache first
    cached = tool_cache.get(tool_name, params)
    if cached is not None:
        return {"result": cached, "cached": True}

    # Execute the tool
    result = tool_registry[tool_name].execute(**params)

    # Cache if the tool is marked as cacheable
    if tool_registry[tool_name].cacheable:
        tool_cache.set(tool_name, params, result)

    return {"result": result, "cached": False}

Performance Metrics

You cannot optimize what you do not measure. Track these four metrics for every agent in production:

Core Metrics Dashboard

import time
from dataclasses import dataclass, field
from collections import defaultdict
from statistics import mean, quantiles

@dataclass
class PerformanceTracker:
    """Track agent performance metrics with percentile calculations."""
    metrics: dict = field(default_factory=lambda: defaultdict(list))

    def record_request(
        self,
        agent_id: str,
        tokens_in: int,
        tokens_out: int,
        latency_ms: float,
        success: bool,
        cost_usd: float,
        business_unit: str,
    ):
        self.metrics[agent_id].append({
            "tokens_total": tokens_in + tokens_out,
            "latency_ms": latency_ms,
            "success": success,
            "cost_usd": cost_usd,
            "business_unit": business_unit,
            "timestamp": time.time(),
        })

    def summary(self, agent_id: str) -> dict:
        records = self.metrics[agent_id]
        if not records:
            return {}

        latencies = [r["latency_ms"] for r in records]
        p50, p95, p99 = quantiles(latencies, n=100)[49], \
                        quantiles(latencies, n=100)[94], \
                        quantiles(latencies, n=100)[98]

        return {
            "avg_tokens_per_request": mean(r["tokens_total"] for r in records),
            "latency_p50_ms": p50,
            "latency_p95_ms": p95,
            "latency_p99_ms": p99,
            "success_rate": mean(r["success"] for r in records),
            "total_cost_usd": sum(r["cost_usd"] for r in records),
            "request_count": len(records),
        }

    def cost_by_business_unit(self, agent_id: str) -> dict:
        """Granular cost attribution for governance reporting."""
        records = self.metrics[agent_id]
        costs = defaultdict(float)
        for r in records:
            costs[r["business_unit"]] += r["cost_usd"]
        return dict(costs)

Setting Targets

Metrics without targets are just numbers. Set thresholds based on your use case:

Cost Tracking with Business Unit Attribution

The governance guide mandates granular cost attribution — every token spent must trace back to a business unit, team, or project. Without this, cost overruns are invisible until the monthly bill arrives.

from dataclasses import dataclass
from datetime import datetime, date
from collections import defaultdict

@dataclass
class CostRecord:
    """Individual cost record for attribution."""
    timestamp: datetime
    agent_id: str
    operation: str
    business_unit: str
    team: str
    tokens_in: int
    tokens_out: int
    model: str
    cost_usd: float

class CostTracker:
    """Granular cost tracking with attribution to business units."""

    # Pricing per million tokens (example rates)
    PRICING = {
        "claude-sonnet": {"input": 3.00, "output": 15.00},
        "claude-haiku": {"input": 0.25, "output": 1.25},
        "gpt-4o": {"input": 2.50, "output": 10.00},
    }

    def __init__(self):
        self.records: list[CostRecord] = []

    def record(self, agent_id: str, operation: str, business_unit: str,
               team: str, model: str, tokens_in: int, tokens_out: int):
        pricing = self.PRICING[model]
        cost = (tokens_in * pricing["input"] + tokens_out * pricing["output"]) / 1_000_000

        self.records.append(CostRecord(
            timestamp=datetime.now(),
            agent_id=agent_id,
            operation=operation,
            business_unit=business_unit,
            team=team,
            tokens_in=tokens_in,
            tokens_out=tokens_out,
            model=model,
            cost_usd=cost,
        ))

    def report_by_business_unit(self, start: date, end: date) -> dict:
        """Generate cost report grouped by business unit."""
        filtered = [r for r in self.records if start <= r.timestamp.date() <= end]
        report = defaultdict(lambda: {"cost": 0.0, "requests": 0, "tokens": 0})

        for r in filtered:
            bu = report[r.business_unit]
            bu["cost"] += r.cost_usd
            bu["requests"] += 1
            bu["tokens"] += r.tokens_in + r.tokens_out

        return dict(report)

    def alert_budget_exceeded(self, budgets: dict[str, float]) -> list[str]:
        """Check if any business unit exceeded their daily budget."""
        today = date.today()
        today_report = self.report_by_business_unit(today, today)
        alerts = []
        for bu, budget in budgets.items():
            if bu in today_report and today_report[bu]["cost"] > budget:
                alerts.append(
                    f"{bu}: ${today_report[bu]['cost']:.2f} exceeds "
                    f"daily budget ${budget:.2f}"
                )
        return alerts

Optimization in Practice

Combining these strategies produces compounding gains. Here's a before-and-after for a code review agent:

The optimizations applied: concise output schema (-28% tokens), prompt caching for system prefix (-50% latency on cache hit), lru_cache on file content lookups (eliminated 34% of API calls), and removal of redundant tool-use instructions (-15% system prompt tokens).

Verify Your Understanding

Apply It

Pick an agent or skill you've built. Measure its current token usage per request (check your API dashboard or add logging). Then apply two optimizations from this lesson: tighten the schema and remove redundant instructions. Measure again. If you achieve less than 15% reduction, look at whether conversation history compression or tool output caching would help — these often deliver larger gains for multi-turn agents.