Observability, Tracing and Cost Control

An AI agent decides what to do at run time, so you cannot read the code to know what it did. Observability means recording enough information (traces, logs and metrics) to answer “what happened, why, how long did it take and how much did it cost?” after the fact. Cost control means putting hard limits around an agent so a bug, a loop or an attack cannot spend your money without end.

Overview: why agents need observability

A normal program follows the code you wrote. An agent runs a loop: it asks a model what to do, calls a tool, reads the result, and asks the model again. The number of loops, the tools chosen and the text sent to the model can change on every run. When an answer is wrong, slow or expensive, you need a record of every step.

Three kinds of data work together:

• Traces show one run from start to end. A trace is a tree of spans. A span is one timed unit of work, such as one model call or one tool call. Each span has a name, a start time, a duration, a parent span and attributes (key-value details).
• Logs are individual events, such as “tool X failed”. Structured logs (one JSON object per line) are much easier to search than free text.
• Metrics are numbers aggregated over many runs: average cost per run, error rate, 95th-percentile latency, tokens per day.

What to record for an agent

For every model call record the model name, input tokens, output tokens, latency, and the stop reason. For every tool call record the tool name, the argument names, success or failure, and latency. For every run record a unique trace_id, the number of steps, total tokens, total cost, and the final status (done, error, step limit, budget limit).

Why cost explodes in agents

Each loop iteration usually re-sends the whole conversation, so input tokens grow with every step. A run with 10 steps can cost far more than ten times a single call. Tool outputs that return large documents make it worse. A bug that makes the agent retry forever can spend money until your provider account runs out. So you need both visibility (see the cost) and limits (stop the cost).

Key concepts

Real systems often use the OpenTelemetry standard or a hosted tracing tool. Their APIs change over time, so check the current documentation of the tool you choose. The ideas below are the same everywhere, and the examples build them from scratch so you see what those tools do internally.

Examples

Example 1: a tiny tracer

This tracer records spans as a tree. A fake clock makes the output repeatable; in real code you would use time.perf_counter.

from contextlib import contextmanager


class FakeClock:
    '''Deterministic clock: each call moves time forward by 0.5 s.'''

    def __init__(self):
        self.now = 0.0

    def __call__(self):
        self.now += 0.5
        return self.now


class Tracer:
    def __init__(self, clock):
        self.clock = clock
        self.spans = []
        self._stack = []

    @contextmanager
    def span(self, name, **attrs):
        record = {
            'id': len(self.spans) + 1,
            'parent': self._stack[-1] if self._stack else None,
            'name': name,
            'attrs': attrs,
            'start': self.clock(),
        }
        self.spans.append(record)
        self._stack.append(record['id'])
        try:
            yield record
        except Exception as exc:
            record['error'] = repr(exc)
            raise
        finally:
            record['end'] = self.clock()
            record['duration'] = record['end'] - record['start']
            self._stack.pop()


tracer = Tracer(FakeClock())
with tracer.span('agent.run', task='weather'):
    with tracer.span('llm.call', step=1) as s:
        s['attrs']['tokens_in'] = 120
    with tracer.span('tool.call', tool='get_weather'):
        pass

for s in tracer.spans:
    print(s['id'], 'parent=' + str(s['parent']), s['name'], str(s['duration']) + 's')

Output:

1 parent=None agent.run 2.0s
2 parent=1 llm.call 0.5s
3 parent=1 tool.call 0.5s

Wait: check the clock. The run span starts at 0.5, the model span runs 1.0 to 1.5, the tool span runs 2.0 to 2.5, and the run span ends at 3.0. So the run lasts 2.5 s, and the correct output is shown below.

1 parent=None agent.run 2.5s
2 parent=1 llm.call 0.5s
3 parent=1 tool.call 0.5s

The _stack list is how a span finds its parent: the span on top of the stack when a new span opens is its parent. The finally block guarantees the end time is recorded even if the code inside fails, and the except block stores the error on the span before re-raising it. Both the llm and tool spans have parent 1, so you can see they ran inside the same run.

Example 2: counting cost and enforcing a budget

Providers bill per token, with different prices for input and output. The prices below are placeholders for illustration only. Real prices change often, so always read them from your provider’s current pricing page.

from dataclasses import dataclass

# Placeholder USD prices per 1 million tokens: (input, output).
PRICES = {'small-model': (0.50, 1.50)}


class BudgetExceeded(Exception):
    pass


@dataclass
class CostTracker:
    budget_usd: float
    spent_usd: float = 0.0
    calls: int = 0

    def record(self, model, tokens_in, tokens_out):
        in_price, out_price = PRICES[model]
        cost = (tokens_in * in_price + tokens_out * out_price) / 1_000_000
        self.spent_usd += cost
        self.calls += 1
        if self.spent_usd > self.budget_usd:
            raise BudgetExceeded(
                f'budget ${self.budget_usd:.4f} exceeded after '
                f'{self.calls} calls (spent ${self.spent_usd:.4f})'
            )
        return cost


tracker = CostTracker(budget_usd=0.01)
for i in range(1, 6):
    try:
        cost = tracker.record('small-model', 4000, 1000)
        print(f'call {i}: cost=${cost:.4f} total=${tracker.spent_usd:.4f}')
    except BudgetExceeded as exc:
        print('stopped:', exc)
        break

Output:

call 1: cost=$0.0035 total=$0.0035
call 2: cost=$0.0035 total=$0.0070
stopped: budget $0.0100 exceeded after 3 calls (spent $0.0105)

Each call costs (4000 x 0.50 + 1000 x 1.50) / 1,000,000 = 0.0035. After the third call the total is 0.0105, which is over 0.01, so the tracker raises an exception and the loop stops. Note that the check happens after a call is made, so you can overshoot by one call. For a stricter guard, estimate the cost of the next call before making it.

Example 3: an agent loop with structured logs, limits and redaction

This example combines everything. The call_llm() function is a stub that returns scripted replies, so the code runs without a network. To use a real provider, replace its body with your SDK call and map the response to the same dictionary shape (type, text or tool name and arguments, and token usage).

import json
import re


def call_llm(messages):
    '''Stub model. Replace the body with a real SDK call.'''
    if any(m['role'] == 'tool' for m in messages):
        return {'type': 'final', 'text': 'It is 24C in Pune.',
                'usage': {'in': 180, 'out': 12}}
    return {'type': 'tool', 'name': 'get_weather', 'args': {'city': 'Pune'},
            'usage': {'in': 95, 'out': 20}}


def get_weather(city):
    return {'city': city, 'temp_c': 24}


TOOLS = {'get_weather': get_weather}


def redact(text):
    '''Hide email addresses before text reaches a log.'''
    return re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[email]', text)


def run_agent(task, max_steps=5):
    trace_id = 'trace-001'
    events = [{'trace': trace_id, 'event': 'start', 'task': redact(task)}]
    messages = [{'role': 'user', 'content': task}]
    total_in = total_out = 0
    for step in range(1, max_steps + 1):
        reply = call_llm(messages)
        total_in += reply['usage']['in']
        total_out += reply['usage']['out']
        events.append({'trace': trace_id, 'step': step, 'event': 'llm',
                       'tokens_in': reply['usage']['in'],
                       'tokens_out': reply['usage']['out']})
        if reply['type'] == 'final':
            events.append({'trace': trace_id, 'event': 'final',
                           'total_in': total_in, 'total_out': total_out})
            return reply['text'], events
        result = TOOLS[reply['name']](**reply['args'])
        events.append({'trace': trace_id, 'step': step, 'event': 'tool',
                       'tool': reply['name'], 'args_keys': list(reply['args'])})
        messages.append({'role': 'tool', 'content': json.dumps(result)})
    raise RuntimeError('max_steps reached')


answer, events = run_agent('Weather in Pune? Mail asha@example.com')
for event in events:
    print(json.dumps(event))
print('Answer:', answer)

Output:

{"trace": "trace-001", "event": "start", "task": "Weather in Pune? Mail [email]"}
{"trace": "trace-001", "step": 1, "event": "llm", "tokens_in": 95, "tokens_out": 20}
{"trace": "trace-001", "step": 1, "event": "tool", "tool": "get_weather", "args_keys": ["city"]}
{"trace": "trace-001", "step": 2, "event": "llm", "tokens_in": 180, "tokens_out": 12}
{"trace": "trace-001", "event": "final", "total_in": 275, "total_out": 32}
Answer: It is 24C in Pune.

Notice four design choices. Every event carries the same trace_id, so you can find a whole run with one search. The email address is redacted before logging. Tool arguments are logged as key names only, not values, because values may be private. And the for loop has a hard max_steps, so the agent cannot loop forever; it raises an error instead.

A single log line looks like this when you ship it to a log system:

{"trace": "trace-001", "step": 2, "event": "llm", "tokens_in": 180, "tokens_out": 12, "model": "small-model", "latency_ms": 840}

Under the hood: what a traced run looks like

1. The run starts and a new trace_id is created.
2. A root span opens for the whole run. Its attributes hold the task id and user id (never raw secrets).
3. Each model call opens a child span. When it ends, you add token counts from the provider’s usage field. Use the numbers the provider returns, not your own guess, because they match the bill.
4. Each tool call opens another child span. Failures are stored as span errors.
5. The cost tracker adds up the cost after every model call and checks the budget and the step count.
6. When the run ends, the root span is closed with a status, and the spans are sent to a backend where you can view them as a timeline.

If the trace shows step 7 took 40 seconds, you know which tool to fix. If it shows input tokens doubling each step, you know the history is growing and needs trimming or summarising.

Common Mistakes

Mistake 1: no limit on the loop

while True:
    reply = call_llm(messages)
    if reply['type'] == 'final':
        break
    messages.append(run_tool(reply))

Why it is wrong: if the model keeps asking for tools (or a tool keeps failing and the model keeps retrying), this never ends and keeps billing. Fix: use for step in range(max_steps) as in Example 3, add a budget check, and add a wall-clock timeout.

Mistake 2: logging everything raw

logger.info('request: %s', json.dumps(full_prompt_with_user_data))
logger.info('tool args: %s', tool_args_with_api_key)

Why it is wrong: prompts and tool arguments can hold passwords, API keys, personal data and text from untrusted web pages. Logs are often readable by many people and stored for a long time. Fix: redact known patterns, log argument names instead of values, truncate long text, and restrict who can read full traces.

Mistake 3: trusting only the final answer

An agent can give a correct answer while wasting ten steps and a lot of money. Always watch cost, step count and latency, not only correctness.

Best Practices

• Give every run a unique trace_id and put it in every span and log line.
• Use structured JSON logs, one event per line.
• Record token usage from the provider’s response for every model call, and compute cost from a price table you keep in config. Prices change, so review it regularly.
• Set limits at several levels: max steps per run, max tokens or money per run, max money per user per day, and a timeout.
• Check the budget before an expensive call when you can estimate its size.
• Redact secrets and personal data before logging. Treat tool output as untrusted text, since it may contain prompt-injection attempts, and flag runs where the agent tried a tool it should not use.
• Log every approval decision when a human approves a risky action, including who approved and when. This gives you an audit trail.
• Reduce cost by trimming old messages, summarising long history, caching repeated results, limiting tool output size, and using a smaller model for easy steps.
• Set alerts on cost per run and error rate, so a sudden spike wakes someone up.
• Sample or expire traces: keep error traces longer, and delete full prompts after a short retention period.

Practice Exercises

1. Extend Tracer so that span() accepts a kind attribute and add a method total_by_kind() that returns the summed duration for each kind. Hint: loop over self.spans and use a dictionary.
2. Change CostTracker so it has a method would_exceed(model, est_in, est_out) that returns True if the next call would break the budget, and use it to stop before the call.
3. Add a second redaction rule to redact() that replaces any text matching sk- followed by letters and digits with [key]. Test it with a fake placeholder string, never a real key.

Summary

• Agents are unpredictable, so you must record traces, structured logs and metrics to understand them.
• A trace is a tree of timed spans linked by a trace id and parent ids.
• Cost grows with every loop step because history is re-sent, so track input and output tokens per call.
• Always enforce limits: max steps, budget, timeout. Never use an unbounded loop.
• Redact secrets and personal data from logs, and treat tool output as untrusted.
• Keep prices in config and check your provider’s pricing page, because prices change.