Production Observability & Cost Control

"If you cannot see what your agent is doing, you cannot afford what your agent is spending."

Sentinel, Cost-Conscious AI Agent

1. Observability for Agentic Systems

Observing what an agent is doing is fundamentally harder than observing a traditional web application. A web request follows a predictable path: receive request, process, respond. An agent loop can take any number of steps, call any combination of tools, and run for seconds or hours. Without proper observability, debugging agent failures becomes a guessing game. "The agent gave a wrong answer" is not actionable; "the agent called the search tool with the wrong query at step 3, retrieved irrelevant documents, and based its answer on those documents" is actionable.

Modern agent observability builds on OpenTelemetry's distributed tracing model. Each agent run is a trace. Each LLM call, tool invocation, and decision point is a span within that trace. Spans capture timing, input/output, token usage, and metadata. This trace structure enables you to see exactly what happened at each step, how long each step took, and what data flowed between steps. Tools like Langfuse, LangSmith, and Arize Phoenix provide pre-built integrations for popular agent frameworks.

The three pillars of agent observability are traces (the complete execution path of an agent run), metrics (aggregated measurements like success rate, average latency, cost per task), and logs (detailed event records for debugging). Traces answer "what happened in this specific run?" Metrics answer "how is the system performing overall?" Logs provide the raw detail needed for root cause analysis when traces point to a problem area.

Langfuse Integration

This snippet instruments an agent pipeline with Langfuse for tracing, logging, and observability.

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe(name="agent_run")
def run_agent(task: str) -> str:
 """Complete agent run with automatic tracing."""

 @observe(name="plan")
 def plan_step(task):
 return llm.invoke(f"Create a plan for: {task}")

 @observe(name="execute_step")
 def execute_step(step, context):
 return agent_executor.invoke(step, context=context)

 @observe(name="synthesize")
 def synthesize(results):
 return llm.invoke(f"Synthesize these results: {results}")

 plan = plan_step(task)
 results = []
 for step in plan.steps:
 result = execute_step(step, results)
 results.append(result)

 return synthesize(results)

# Each call creates spans in the Langfuse trace
# Visible in the Langfuse dashboard with timing, tokens, costs
result = run_agent("Analyze Q3 sales trends")

2. Cost Control and Budget Enforcement

Agent systems can generate unpredictable costs because the number of LLM calls per task varies. A task that should take 3 tool calls might enter a retry loop and make 30 calls before the maximum attempt limit is reached. Without budget enforcement, a single runaway agent can consume a significant portion of the monthly API budget. Implementing cost controls at multiple levels prevents this: per-task budgets, per-user budgets, per-hour rate limits, and system-wide spending caps.

The simplest and most effective cost control is a per-task token budget. Before the agent starts, calculate the expected token cost based on the task type and set a hard limit at 3 to 5 times that expectation. If the agent exceeds the budget, it must produce the best answer it can with the remaining tokens or terminate gracefully with a partial result. This prevents runaway costs while allowing headroom for tasks that genuinely need more processing.

class BudgetEnforcer:
 def __init__(self, max_tokens: int, max_cost_usd: float):
 self.max_tokens = max_tokens
 self.max_cost_usd = max_cost_usd
 self.used_tokens = 0
 self.used_cost = 0.0

 def check_budget(self, estimated_tokens: int) -> bool:
 """Check if the next LLM call is within budget."""
 if self.used_tokens + estimated_tokens > self.max_tokens:
 return False
 return self.used_cost < self.max_cost_usd

 def record_usage(self, input_tokens: int, output_tokens: int, model: str):
 """Record token usage and update cost tracking."""
 self.used_tokens += input_tokens + output_tokens
 cost = calculate_cost(input_tokens, output_tokens, model)
 self.used_cost += cost

 def remaining_budget(self) -> dict:
 return {
 "tokens_remaining": self.max_tokens - self.used_tokens,
 "cost_remaining": self.max_cost_usd - self.used_cost,
 "utilization": self.used_cost / self.max_cost_usd,
 }

3. Alerting and Anomaly Detection

Proactive alerting catches agent issues before they impact users. Set up alerts for: task failure rate exceeding a threshold (e.g., >10% failures in a 5-minute window), average latency exceeding SLA targets, cost per task spiking above baseline, tool call error rates increasing (may indicate an upstream API issue), and agent loop count exceeding expected bounds (the agent is stuck in a retry loop).

Anomaly detection goes beyond static thresholds by learning the normal behavior patterns of your agent system. A model-based anomaly detector can flag when the distribution of tool calls changes (the agent is suddenly calling a tool it rarely uses), when response times shift (latency increased but no code changed, suggesting a provider issue), or when output quality degrades (detected through automated quality checks or increased user complaints). Time-series anomaly detection using simple statistical methods (z-score, moving average) works well for most agent monitoring use cases.

Exercises