Agent Monitoring & Observability

You Can’t Debug What You Can’t See

Imagine your agent is in production, handling customer requests overnight. A user complains that it gave wrong information. How do you figure out what happened?

Without observability, you’re guessing. With observability, you can replay the exact sequence of LLM calls and tool results from that user’s session, see exactly where the reasoning went wrong, and fix it.

Observability for agents means complete visibility into:

• What the agent decided to do and why (reasoning traces)
• What tools it called, with what arguments, and what they returned
• How long each step took and how many tokens it consumed
• What the final output was and whether it was correct

This module covers practical tools and patterns for achieving this visibility.

Structured Logging for Every Agent Action

The foundation of observability is structured logging, writing machine-readable JSON logs rather than plain text strings. These can be aggregated, queried, and visualised.

Here’s a logging wrapper for all agent actions:

import json, time, uuid
from datetime import datetime

def create_run_id() -> str:
    return str(uuid.uuid4())[:8]

def log_event(run_id: str, event_type: str, data: dict):
    """Write a structured log event to stdout (or your log aggregator)."""
    entry = {
        'timestamp': datetime.utcnow().isoformat() + 'Z',
        'run_id':    run_id,
        'event':     event_type,
        **data
    }
    print(json.dumps(entry))

Wrap every LLM call to log tokens and latency:

from openai import OpenAI
client = OpenAI()

def logged_llm_call(run_id: str, messages: list, **kwargs) -> str:
    """Call the LLM and log detailed metrics."""
    start = time.time()
    
    response = client.chat.completions.create(messages=messages, **kwargs)
    latency_ms = (time.time() - start) * 1000
    usage = response.usage
    
    log_event(run_id, 'llm_call', {
        'model':          kwargs.get('model', 'unknown'),
        'input_tokens':   usage.prompt_tokens,
        'output_tokens':  usage.completion_tokens,
        'latency_ms':     round(latency_ms, 1),
        'finish_reason':  response.choices[0].finish_reason,
    })
    
    return response.choices[0].message.content

Log tool calls the same way:

def logged_tool_call(run_id: str, name: str, args: dict) -> str:
    """Execute a tool and log what happened."""
    start = time.time()
    
    try:
        result = TOOL_REGISTRY[name](**args)
        latency_ms = (time.time() - start) * 1000
        
        log_event(run_id, 'tool_call', {
            'tool':       name,
            'args':       args,
            'success':    True,
            'result_len': len(str(result)),
            'latency_ms': round(latency_ms, 1),
        })
        return str(result)
    
    except Exception as e:
        log_event(run_id, 'tool_error', {'tool': name, 'args': args, 'error': str(e)})
        return f"Error: {e}"

Cost Tracking Per Run

Add up token costs in real-time so you can catch runaway agents before they exhaust your budget:

# OpenAI pricing (always verify current rates at platform.openai.com/pricing)
TOKEN_COSTS = {
    'gpt-5.6-sol':      {'input': 2.50, 'output': 10.00},   # per million tokens
    'gpt-5.6-luna': {'input': 0.15, 'output':  0.60},
}

class CostTracker:
    def __init__(self, budget_usd: float = 1.0):
        self.total_cost = 0.0
        self.budget     = budget_usd
        self.llm_calls  = 0
    
    def add(self, model: str, input_tokens: int, output_tokens: int):
        rates = TOKEN_COSTS.get(model, {'input': 5.0, 'output': 15.0})
        cost  = (input_tokens * rates['input'] + output_tokens * rates['output']) / 1_000_000
        
        self.total_cost += cost
        self.llm_calls  += 1
        
        if self.total_cost > self.budget:
            raise RuntimeError(
                f"Budget exceeded: ${self.total_cost:.4f} > ${self.budget:.2f}"
            )
    
    def summary(self) -> dict:
        return {
            'total_usd': round(self.total_cost, 4),
            'budget_usd': self.budget,
            'llm_calls': self.llm_calls,
        }

Use it in your agent loop:

def run_agent_with_budget(goal: str, budget_usd: float = 0.50) -> dict:
    run_id  = create_run_id()
    tracker = CostTracker(budget_usd=budget_usd)
    
    log_event(run_id, 'run_start', {'goal': goal[:200], 'budget': budget_usd})
    
    messages = [
        {'role': 'system', 'content': AGENT_SYSTEM_PROMPT},
        {'role': 'user',   'content': goal},
    ]
    
    try:
        for step in range(20):
            response = client.chat.completions.create(
                model='gpt-5.6-sol', messages=messages, tools=tools
            )
            usage = response.usage
            tracker.add('gpt-5.6-sol', usage.prompt_tokens, usage.completion_tokens)
            
            msg = response.choices[0].message
            messages.append(msg)
            
            if not msg.tool_calls:
                log_event(run_id, 'run_end', tracker.summary())
                return {'result': msg.content, 'cost': tracker.summary()}
            
            # handle tool calls...
    
    except RuntimeError as e:
        log_event(run_id, 'budget_exceeded', {'error': str(e)})
        return {'error': str(e), 'cost': tracker.summary()}

LangSmith: Full Trace Visualisation

LangSmith is LangChain’s hosted observability platform that captures full traces from LangChain and LangGraph agents automatically. Setup takes under 5 minutes:

pip install langsmith
export LANGCHAIN_API_KEY="your-langsmith-api-key"
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_PROJECT="my-agent-project"

With these environment variables set, any LangChain/LangGraph code automatically sends traces to LangSmith, no code changes required. Visit smith.langchain.com to see:

• Full timeline of every LLM call and tool call
• Input and output for each step with token counts and latency
• Side-by-side comparisons of different runs
• Automated evaluation scores

For non-LangChain agents, use the @traceable decorator:

from langsmith.run_helpers import traceable

@traceable(name="research-agent", run_type="chain")
def traced_agent(goal: str) -> str:
    """This function's execution is automatically traced in LangSmith."""
    # ... your agent implementation ...
    return result

# When called, this entire execution shows up as one trace
result = traced_agent("Research the latest LLM benchmarks")

Alerting on Anomalies

Define thresholds and alert when agents misbehave:

THRESHOLDS = {
    'max_steps':     20,
    'max_cost_usd':  1.0,
    'max_latency_s': 60,
}

def check_and_alert(run_summary: dict):
    alerts = []
    
    if run_summary.get('llm_calls', 0) > THRESHOLDS['max_steps']:
        alerts.append(f"Steps exceeded: {run_summary['llm_calls']}")
    
    if run_summary.get('total_usd', 0) > THRESHOLDS['max_cost_usd']:
        alerts.append(f"Cost exceeded: ${run_summary['total_usd']:.3f}")
    
    if alerts:
        # In production: send to Slack, PagerDuty, email
        for alert in alerts:
            print(f"ALERT [{run_summary.get('run_id')}]: {alert}")

In production, route these alerts to Slack via webhook or to your on-call rotation via PagerDuty. The goal is to catch problems within minutes, not hours.

Exercise: Add structured logging to any agent from a previous module. Run it on 5 different inputs. After each run, print the total token cost and number of LLM calls. Find the most expensive run, what caused it to use more tokens? Is there a way to reduce the cost without sacrificing quality?