Skip to content

AI Agent Reliability: Logging, Error Handling & Monitoring Guide for Production Systems

Problem

When I deployed my first AI agent to production, I got this error:

Agent execution failed. Reason: unknown.

That was it. No stack trace, no context, no way to debug. The agent worked perfectly in development but failed silently in production. I spent three days trying to figure out what went wrong.

Environment

  • Python 3.11
  • LangChain 0.1.x
  • OpenAI GPT-4
  • AWS Lambda (production)

What happened?

My AI agent was processing user requests, calling external APIs, and making decisions based on LLM outputs. Everything looked fine during testing. But in production, failures started accumulating:

Day 1: 47 requests processed, 3 failures (6% failure rate)
Day 2: 89 requests processed, 12 failures (13% failure rate)
Day 3: 156 requests processed, 34 failures (22% failure rate)

The failure rate was climbing, and I had no idea why. My logs showed nothing useful:

agent-bad.py
def run_agent(prompt: str):
result = agent.run(prompt)
return result # That's it. No logging, no error handling.

This is the “slot machine” problem that one Reddit commenter described perfectly: “If you do not know why your agent failed you did not build a system. You built a slot machine.”

The Reliability Crisis

AI agents fail in production for reasons that traditional software doesn’t encounter:

Hallucinations: The LLM makes up facts or generates invalid outputs that break downstream logic. One Reddit commenter put it bluntly: “In production models hallucinate constantly. You cannot trust the output blindly. You need to write code that validates the response.”

Tool failures: External APIs timeout, return unexpected formats, or rate-limit your requests.

Context drift: As conversations get longer, the agent loses track of earlier decisions or contradicts itself.

Token limits: The agent hits context window limits mid-execution, leaving tasks incomplete.

The gap between demo and production is massive. A demo that works 90% of the time is impressive. A production system that fails 10% of the time is unacceptable.

How to solve it?

I rebuilt my agent with a comprehensive observability stack: structured logging, multi-layered error handling, and real-time monitoring. Here’s what that looks like.

Step 1: Structured Logging

First, I added logging to every decision point:

agent-logging.py
import structlog
from datetime import datetime
import uuid
logger = structlog.get_logger()
class AgentLogger:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.session_id = str(uuid.uuid4())
def log_tool_call(self, tool_name: str, args: dict):
logger.info(
"tool_call_started",
agent_id=self.agent_id,
session_id=self.session_id,
tool=tool_name,
arguments=args,
timestamp=datetime.utcnow().isoformat()
)
def log_tool_result(self, tool_name: str, result: str, duration_ms: float):
logger.info(
"tool_call_completed",
agent_id=self.agent_id,
session_id=self.session_id,
tool=tool_name,
result_preview=result[:200],
duration_ms=duration_ms
)
def log_model_decision(self, decision: str, reasoning: str):
logger.info(
"model_decision",
agent_id=self.agent_id,
session_id=self.session_id,
decision=decision,
reasoning=reasoning
)
def log_error(self, error_type: str, message: str, context: dict):
logger.error(
"agent_error",
agent_id=self.agent_id,
session_id=self.session_id,
error_type=error_type,
message=message,
context=context
)

Now every tool call and model decision is logged with correlation IDs. When something fails, I can trace the entire execution path.

Step 2: Multi-Layered Error Handling

Not all errors are equal. I implemented a four-layer error handling strategy:

error-handling.py
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class ToolError:
error_type: str
message: str
recoverable: bool
retry_after: Optional[int] = None
class ErrorHandler:
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
def handle_tool_error(self, error: Exception, tool_name: str) -> ToolError:
# Layer 1: Transient failures - automatic retry
if self._is_transient(error):
return ToolError(
error_type="transient",
message=str(error),
recoverable=True,
retry_after=2
)
# Layer 2: LLM-recoverable errors - return context to model
if self._is_llm_recoverable(error):
return ToolError(
error_type="llm_recoverable",
message=f"Tool {tool_name} failed: {error}. Try with different parameters.",
recoverable=True
)
# Layer 3: User-fixable problems - prompt for intervention
if self._is_user_fixable(error):
return ToolError(
error_type="user_fixable",
message=f"This action requires your input: {error}",
recoverable=True
)
# Layer 4: Unexpected errors - bubble up for alerting
return ToolError(
error_type="unexpected",
message=str(error),
recoverable=False
)
def _is_transient(self, error: Exception) -> bool:
transient_errors = [
"Rate limit exceeded",
"Connection timeout",
"Service temporarily unavailable"
]
return any(te in str(error) for te in transient_errors)
def _is_llm_recoverable(self, error: Exception) -> bool:
llm_recoverable_errors = [
"Invalid parameter",
"Missing required field",
"Format error"
]
return any(le in str(error) for le in llm_recoverable_errors)
def _is_user_fixable(self, error: Exception) -> bool:
user_fixable_errors = [
"Authentication required",
"Permission denied",
"Resource not found"
]
return any(ue in str(error) for ue in user_fixable_errors)
def execute_with_retry(tool, args: dict, handler: ErrorHandler, logger: AgentLogger):
for attempt in range(handler.max_retries):
try:
start_time = time.time()
logger.log_tool_call(tool.name, args)
result = tool.run(args)
duration = (time.time() - start_time) * 1000
logger.log_tool_result(tool.name, result, duration)
return result
except Exception as e:
error_info = handler.handle_tool_error(e, tool.name)
logger.log_error(error_info.error_type, error_info.message, {"attempt": attempt + 1})
if not error_info.recoverable:
raise
if error_info.retry_after:
time.sleep(error_info.retry_after * (2 ** attempt)) # Exponential backoff
elif error_info.error_type == "llm_recoverable":
return error_info.message # Return error context to LLM
elif error_info.error_type == "user_fixable":
return f"[REQUIRES_USER_INPUT] {error_info.message}"
raise Exception(f"Max retries ({handler.max_retries}) exceeded for tool {tool.name}")

Step 3: Output Validation

Never trust model outputs blindly. Here’s how I validate responses:

output-validation.py
from pydantic import BaseModel, ValidationError
from typing import List, Optional
class AgentOutput(BaseModel):
action: str
parameters: dict
reasoning: str
confidence: float
class OutputValidator:
def __init__(self, schema: type[BaseModel]):
self.schema = schema
def validate(self, raw_output: str) -> tuple[bool, Optional[AgentOutput], Optional[str]]:
try:
import json
parsed = json.loads(raw_output)
validated = self.schema.model_validate(parsed)
# Additional semantic validation
if validated.confidence < 0.5:
return False, None, "Confidence too low. Consider gathering more information."
if not validated.reasoning or len(validated.reasoning) < 10:
return False, None, "Reasoning is too short. Provide more context."
return True, validated, None
except json.JSONDecodeError as e:
return False, None, f"Invalid JSON: {e}"
except ValidationError as e:
return False, None, f"Schema validation failed: {e}"
def validate_tool_output(self, tool_name: str, output: str) -> tuple[bool, Optional[str]]:
# Define expected formats per tool
tool_validators = {
"search": self._validate_search_output,
"calculator": self._validate_calculator_output,
"database": self._validate_database_output
}
validator = tool_validators.get(tool_name, self._validate_generic_output)
return validator(output)
def _validate_search_output(self, output: str) -> tuple[bool, Optional[str]]:
if not output or output == "No results found":
return False, "Search returned no results. Try different keywords."
return True, None
def _validate_calculator_output(self, output: str) -> tuple[bool, Optional[str]]:
try:
result = float(output)
if result == float('inf') or result == float('-inf'):
return False, "Calculation resulted in infinity. Check inputs."
return True, None
except ValueError:
return False, f"Calculator output is not a number: {output}"
def _validate_generic_output(self, output: str) -> tuple[bool, Optional[str]]:
if not output or output.strip() == "":
return False, "Tool returned empty output."
return True, None

Step 4: Real-Time Monitoring

I built a monitoring system that streams agent execution events in real-time:

monitoring.py
from dataclasses import dataclass
from datetime import datetime
from typing import Generator, Optional
import asyncio
@dataclass
class AgentEvent:
event_type: str
timestamp: datetime
agent_id: str
session_id: str
data: dict
class AgentMonitor:
def __init__(self):
self.events: list[AgentEvent] = []
self.subscribers: list = []
def record_event(self, event_type: str, agent_id: str, session_id: str, data: dict):
event = AgentEvent(
event_type=event_type,
timestamp=datetime.utcnow(),
agent_id=agent_id,
session_id=session_id,
data=data
)
self.events.append(event)
self._notify_subscribers(event)
def subscribe(self, callback):
self.subscribers.append(callback)
def _notify_subscribers(self, event: AgentEvent):
for callback in self.subscribers:
callback(event)
def get_metrics(self, session_id: str) -> dict:
session_events = [e for e in self.events if e.session_id == session_id]
tool_calls = [e for e in session_events if e.event_type == "tool_call"]
errors = [e for e in session_events if e.event_type == "error"]
durations = [e.data.get("duration_ms", 0) for e in tool_calls if "duration_ms" in e.data]
return {
"total_tool_calls": len(tool_calls),
"total_errors": len(errors),
"avg_tool_duration_ms": sum(durations) / len(durations) if durations else 0,
"error_rate": len(errors) / len(tool_calls) if tool_calls else 0
}
async def run_agent_with_monitoring(agent, prompt: str, monitor: AgentMonitor):
session_id = str(uuid.uuid4())
monitor.record_event("session_started", agent.id, session_id, {"prompt": prompt})
try:
result = await agent.arun(prompt)
monitor.record_event("session_completed", agent.id, session_id, {"result": result})
return result
except Exception as e:
monitor.record_event("error", agent.id, session_id, {"error": str(e), "type": type(e).__name__})
raise

The Architecture

Here’s how the complete observability stack fits together:

┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Request │────▶│ Agent │────▶│ Tools │
└─────────────┘ │ (with │ │ │
│ logging) │ └─────────────┘
└─────────────┘ │
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Error │ │ Output │
│ Handler │ │ Validator │
└─────────────┘ └─────────────┘
│ │
└──────────┬─────────┘
┌─────────────┐
│ Monitor │
│ (events) │
└─────────────┘
┌─────────────┐
│ Storage │
│ (SQLite) │
└─────────────┘

Real Results

After implementing this observability stack, my production metrics changed dramatically:

Before:
- Failure rate: 22%
- Mean time to diagnose: 3+ days
- Cost per task: Unknown
- Recovery rate: 0%
After:
- Failure rate: 2.1%
- Mean time to diagnose: < 30 minutes
- Cost per task: $0.03
- Recovery rate: 87%

The key improvement was visibility. When a failure occurred, I could see exactly which tool failed, what arguments were passed, and what the error was. The error handling middleware automatically recovered 87% of failures without human intervention.

Production Metrics That Matter

I track these metrics for every agent in production:

Reliability Metrics:

  • Success rate per task type
  • Error rate by error category
  • Recovery rate (automatic vs manual)
  • Mean time to recovery

Performance Metrics:

  • Step-wise latency (time per tool call)
  • Total execution time
  • Token usage per task
  • API call frequency

Cost Metrics:

  • Cost per successful task
  • Cost per failed task
  • Expensive tool identification
  • Budget burn rate

Quality Metrics:

  • Output validation pass rate
  • User satisfaction signals
  • Task completion rate
  • Re-execution rate (tasks that needed retry)

Anti-Patterns to Avoid

I learned these lessons the hard way:

Skipping logging until later: It never happens. Build logging from day one.

Trusting model outputs: Always validate structure, content, and logic before using outputs.

Generic error messages: “Something went wrong” is useless. Include context, error type, and recovery suggestions.

Siloed monitoring: Use correlation IDs to trace requests across all components.

Ignoring costs until budget overruns: Token costs add up fast. Monitor them continuously.

Implementation Roadmap

If you’re starting from scratch, here’s the order I recommend:

  1. Add structured logging to all tool calls and agent decisions
  2. Implement error handling middleware with custom error types
  3. Set up real-time event streaming and dashboards
  4. Add cost and performance monitoring
  5. Implement output validation and guardrails
  6. Set up alerting and automated failure recovery

Summary

In this post, I showed how to build reliable AI agents with proper logging, error handling, and monitoring. The key insight is that reliability is the only metric that matters for production AI agents. As one Reddit commenter said: “Clients do not care if you used the latest model. They only care if the system runs every single day without breaking.”

The observability stack I described transforms AI agents from slot machines into reliable systems:

  • Structured logging provides the debugging foundation
  • Multi-layered error handling recovers gracefully from failures
  • Output validation catches hallucinations before they impact users
  • Real-time monitoring gives visibility into agent behavior
  • Cost tracking prevents budget surprises

Without observability, AI agents are unpredictable prototypes. With it, they become production-ready systems that clients can depend on.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments