Introduction to AI Agent Monitoring
When AI agents move from development notebooks into production, they become autonomous software components that interact with users, APIs, and other systems. Monitoring these agents is not a luxury ā it's a critical operational necessity. Unlike traditional deterministic software, AI agents introduce non-deterministic behavior, variable latency, token-based pricing models, and complex failure modes that require specialized observability strategies.
This tutorial walks you through building a complete monitoring pipeline for AI agents in production, covering structured logging, cost tracking, and error handling with practical code examples you can integrate immediately.
What Is AI Agent Monitoring?
AI agent monitoring is the practice of instrumenting autonomous LLM-powered agents to capture telemetry data across three primary dimensions:
- Execution logs ā full traces of agent reasoning, tool calls, and final outputs
- Cost metrics ā token consumption, model usage, and cumulative spend per session or user
- Error tracking ā model refusals, tool failures, retry exhaustion, and guardrail violations
Together, these signals give you observability into agent performance, reliability, and financial impact ā enabling you to debug failures, optimize prompts, and control runaway costs before they surprise your finance team.
Why Monitoring Matters for Production Agents
Traditional application monitoring assumes predictable execution paths and binary pass/fail outcomes. AI agents break both assumptions. An agent may produce a correct answer through an unexpected reasoning chain, or fail silently by returning a plausible but wrong result. Costs are usage-based and can spike unpredictably if an agent enters a reasoning loop or invokes expensive tools repeatedly. Without purpose-built monitoring, you lose visibility into:
- Whether the agent actually followed its intended decision process
- How many tokens each step consumed and what the cumulative cost was
- Which tool calls failed and why the agent didn't recover
- Patterns of user sessions that lead to high-cost or high-error outcomes
Monitoring bridges the gap between the black-box behavior of LLMs and the operational requirements of production systems.
Core Pillar 1: Structured Logging for Agent Traces
What to Log
Unlike flat application logs, agent logging requires capturing hierarchical, multi-step traces. A typical agent invocation includes a planning phase, multiple reasoning-and-tool-calling cycles, and a final synthesis step. Each of these must be logged with enough context to reconstruct the agent's decision-making path during post-mortem analysis.
At minimum, your agent logger should capture:
- A unique trace ID and span ID for each invocation and sub-step
- The user input and any session context
- Each LLM call with the full prompt (or a hash), the response, and token counts
- Every tool invocation with its name, input arguments, output, and duration
- Intermediate reasoning steps if your agent framework exposes them
- The final output delivered to the user
Implementation Example with OpenTelemetry
Below is a production-ready logging setup using Python, OpenTelemetry, and a simple agent loop. The code demonstrates how to wrap LLM calls and tool executions in spans that preserve parent-child relationships.
# agent_monitor.py
import time
import uuid
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
# Configure OpenTelemetry
resource = Resource.create({ResourceAttributes.SERVICE_NAME: "ai-agent-runtime"})
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)
otlp_exporter = OTLPSpanExporter(
endpoint="http://localhost:4317",
insecure=True
)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
tracer = trace.get_tracer(__name__)
class AgentLogger:
"""Structured logger for multi-step agent executions."""
def __init__(self):
self.active_traces = {}
def start_invocation(self, user_input: str, session_id: str = None) -> str:
trace_id = str(uuid.uuid4())
span = tracer.start_span(
name="agent_invocation",
attributes={
"agent.trace_id": trace_id,
"agent.session_id": session_id or str(uuid.uuid4()),
"agent.user_input": user_input[:500], # truncate long inputs
"agent.start_time": time.time()
}
)
self.active_traces[trace_id] = {
"root_span": span,
"step_count": 0
}
return trace_id
def log_llm_call(self, trace_id: str, step: int,
model: str, prompt: str, response: str,
prompt_tokens: int, completion_tokens: int, latency_ms: float):
"""Log an individual LLM completion within an agent trace."""
if trace_id not in self.active_traces:
raise ValueError(f"Unknown trace_id: {trace_id}")
root_span = self.active_traces[trace_id]["root_span"]
ctx = trace.set_span_in_context(root_span)
with tracer.start_as_current_span(
name="agent.llm_call",
context=ctx,
attributes={
"agent.step": step,
"agent.model": model,
"agent.prompt_hash": hash(prompt),
"agent.prompt_tokens": prompt_tokens,
"agent.completion_tokens": completion_tokens,
"agent.total_tokens": prompt_tokens + completion_tokens,
"agent.latency_ms": latency_ms,
"agent.response_preview": response[:200]
}
) as llm_span:
# Store full prompt and response as span events for detailed debugging
llm_span.add_event("prompt", {"content": prompt})
llm_span.add_event("response", {"content": response})
self.active_traces[trace_id]["step_count"] = step
def log_tool_call(self, trace_id: str, step: int,
tool_name: str, tool_input: dict, tool_output: dict,
success: bool, duration_ms: float, error_message: str = None):
"""Log a tool execution within an agent trace."""
if trace_id not in self.active_traces:
raise ValueError(f"Unknown trace_id: {trace_id}")
root_span = self.active_traces[trace_id]["root_span"]
ctx = trace.set_span_in_context(root_span)
with tracer.start_as_current_span(
name=f"agent.tool_call.{tool_name}",
context=ctx,
attributes={
"agent.step": step,
"agent.tool_name": tool_name,
"agent.tool_input": str(tool_input),
"agent.tool_output": str(tool_output),
"agent.tool_success": success,
"agent.tool_duration_ms": duration_ms,
"agent.tool_error": error_message or ""
}
) as tool_span:
if error_message:
tool_span.set_status(trace.Status(trace.StatusCode.ERROR, error_message))
def finish_invocation(self, trace_id: str, final_output: str,
total_cost: float, total_latency_ms: float):
"""Close the root span and export the complete trace."""
if trace_id not in self.active_traces:
raise ValueError(f"Unknown trace_id: {trace_id}")
root_span = self.active_traces[trace_id]["root_span"]
root_span.set_attribute("agent.final_output", final_output[:1000])
root_span.set_attribute("agent.total_cost_usd", total_cost)
root_span.set_attribute("agent.total_latency_ms", total_latency_ms)
root_span.set_attribute("agent.total_steps",
self.active_traces[trace_id]["step_count"])
root_span.end()
del self.active_traces[trace_id]
return trace_id
# Usage example
logger = AgentLogger()
trace_id = logger.start_invocation(
"Summarize the latest quarterly report and email it to the CFO",
session_id="user-42-session-abc"
)
# Simulate agent loop
logger.log_llm_call(
trace_id, step=1, model="gpt-4o",
prompt="You are an executive assistant...",
response="I will first retrieve the report, then compose a summary...",
prompt_tokens=450, completion_tokens=120, latency_ms=1850
)
logger.log_tool_call(
trace_id, step=2, tool_name="fetch_report",
tool_input={"quarter": "Q3", "year": 2024},
tool_output={"report_id": "rpt-9921", "size_bytes": 45120},
success=True, duration_ms=340
)
logger.log_llm_call(
trace_id, step=3, model="gpt-4o",
prompt="Here is the report data... Summarize it in 3 bullet points.",
response="ā¢ Revenue grew 12% YoY...\nā¢ Operating margin expanded...",
prompt_tokens=1200, completion_tokens=300, latency_ms=4100
)
logger.finish_invocation(
trace_id,
final_output="Revenue grew 12% YoY...",
total_cost=0.042,
total_latency_ms=6290
)Visualizing Agent Traces
Once traces are exported via OTLP, you can use Jaeger, Grafana Tempo, or Datadog to visualize them. The hierarchical span structure lets you drill down into any agent invocation and see exactly which LLM calls and tool executions occurred, in what order, with full context. This is invaluable for debugging unexpected agent behavior ā you can replay the exact sequence of prompts and tool responses that led to a particular outcome.
Core Pillar 2: Cost Tracking
Why Cost Tracking Is Essential
AI agent costs are inherently variable. A single user query might cost $0.01 or $2.50 depending on the complexity of the reasoning chain, the number of tool calls, and the model tier invoked. Without cost instrumentation, you cannot answer basic operational questions like: "What was our total LLM spend yesterday?", "Which users consume the most expensive agent sessions?", or "Is our new prompt optimization actually reducing costs?"
Cost tracking must be real-time and granular ā attributed to sessions, users, and individual agent steps ā so you can set budgets, trigger alerts, and optimize high-cost paths.
Building a Cost Calculator
The following implementation tracks per-model pricing and computes cumulative costs across an agent session. It integrates with the logger from the previous section.
# cost_tracker.py
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
@dataclass
class ModelPricing:
"""Pricing per 1M tokens for input and output."""
input_cost_per_million: float
output_cost_per_million: float
# Pricing as of November 2024 ā update regularly
MODEL_PRICING: Dict[str, ModelPricing] = {
"gpt-4o": ModelPricing(input_cost_per_million=2.50, output_cost_per_million=10.00),
"gpt-4o-mini": ModelPricing(input_cost_per_million=0.15, output_cost_per_million=0.60),
"gpt-4-turbo": ModelPricing(input_cost_per_million=10.00, output_cost_per_million=30.00),
"claude-3-opus": ModelPricing(input_cost_per_million=15.00, output_cost_per_million=75.00),
"claude-3-sonnet": ModelPricing(input_cost_per_million=3.00, output_cost_per_million=15.00),
"claude-3-haiku": ModelPricing(input_cost_per_million=0.25, output_cost_per_million=1.25),
}
@dataclass
class CostEntry:
step: int
model: str
prompt_tokens: int
completion_tokens: int
cost_usd: float
timestamp: float = field(default_factory=time.time)
class CostTracker:
"""Tracks cumulative costs per agent session."""
def __init__(self):
self.sessions: Dict[str, List[CostEntry]] = {}
self.session_metadata: Dict[str, dict] = {}
def start_session(self, session_id: str, user_id: str,
feature: str = "default"):
"""Initialize cost tracking for a new session."""
self.sessions[session_id] = []
self.session_metadata[session_id] = {
"user_id": user_id,
"feature": feature,
"start_time": time.time()
}
def record_llm_cost(self, session_id: str, step: int,
model: str, prompt_tokens: int,
completion_tokens: int) -> float:
"""Calculate and record cost for an LLM call."""
if session_id not in self.sessions:
raise ValueError(f"Unknown session: {session_id}")
pricing = MODEL_PRICING.get(model)
if not pricing:
# Default to a conservative estimate for unknown models
pricing = ModelPricing(input_cost_per_million=5.0,
output_cost_per_million=20.0)
input_cost = (prompt_tokens / 1_000_000) * pricing.input_cost_per_million
output_cost = (completion_tokens / 1_000_000) * pricing.output_cost_per_million
total_cost = round(input_cost + output_cost, 6)
entry = CostEntry(
step=step,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
cost_usd=total_cost
)
self.sessions[session_id].append(entry)
return total_cost
def get_session_total(self, session_id: str) -> float:
"""Get cumulative cost for a session."""
entries = self.sessions.get(session_id, [])
return round(sum(e.cost_usd for e in entries), 6)
def get_session_breakdown(self, session_id: str) -> dict:
"""Get detailed cost breakdown by model and step."""
entries = self.sessions.get(session_id, [])
if not entries:
return {"total": 0.0, "by_model": {}, "steps": []}
by_model = {}
for entry in entries:
if entry.model not in by_model:
by_model[entry.model] = 0.0
by_model[entry.model] += entry.cost_usd
return {
"session_id": session_id,
"total": round(sum(e.cost_usd for e in entries), 6),
"step_count": len(entries),
"by_model": {k: round(v, 6) for k, v in by_model.items()},
"steps": [
{
"step": e.step,
"model": e.model,
"tokens": e.prompt_tokens + e.completion_tokens,
"cost": e.cost_usd
}
for e in entries
],
"metadata": self.session_metadata.get(session_id, {})
}
def get_daily_cost_summary(self, session_ids: List[str]) -> dict:
"""Aggregate costs across multiple sessions for reporting."""
total = 0.0
by_feature = {}
by_model = {}
for sid in session_ids:
breakdown = self.get_session_breakdown(sid)
total += breakdown["total"]
feature = self.session_metadata.get(sid, {}).get("feature", "unknown")
if feature not in by_feature:
by_feature[feature] = 0.0
by_feature[feature] += breakdown["total"]
for model, cost in breakdown.get("by_model", {}).items():
if model not in by_model:
by_model[model] = 0.0
by_model[model] += cost
return {
"total_cost": round(total, 4),
"session_count": len(session_ids),
"average_cost_per_session": round(total / max(len(session_ids), 1), 4),
"by_feature": {k: round(v, 4) for k, v in by_feature.items()},
"by_model": {k: round(v, 4) for k, v in by_model.items()}
}
def export_for_billing(self, session_id: str) -> dict:
"""Export a structured record suitable for billing systems."""
breakdown = self.get_session_breakdown(session_id)
return {
"session_id": session_id,
"timestamp": self.session_metadata.get(session_id, {}).get("start_time"),
"user_id": self.session_metadata.get(session_id, {}).get("user_id"),
"total_cost": breakdown["total"],
"line_items": [
{
"description": f"LLM call step {s['step']} ({s['model']})",
"tokens": s["tokens"],
"cost": s["cost"]
}
for s in breakdown["steps"]
]
}
# Usage example integrated with agent loop
cost_tracker = CostTracker()
session_id = "user-42-session-abc"
cost_tracker.start_session(session_id, user_id="user-42", feature="executive_summaries")
# Step 1 cost
cost1 = cost_tracker.record_llm_cost(
session_id, step=1, model="gpt-4o",
prompt_tokens=450, completion_tokens=120
)
print(f"Step 1 cost: ${cost1:.6f}") # ~$0.002325
# Step 3 cost (after tool call)
cost3 = cost_tracker.record_llm_cost(
session_id, step=3, model="gpt-4o",
prompt_tokens=1200, completion_tokens=300
)
print(f"Step 3 cost: ${cost3:.6f}") # ~$0.006000
total = cost_tracker.get_session_total(session_id)
print(f"Session total: ${total:.6f}") # ~$0.008325
# Export for billing
billing_record = cost_tracker.export_for_billing(session_id)
print(billing_record)Cost Alerting and Budgets
To prevent cost overruns, implement real-time checks that halt or escalate expensive sessions. Here's a lightweight cost guard that integrates with the tracker:
# cost_guard.py
class CostGuard:
"""Enforces cost limits on agent sessions."""
def __init__(self, cost_tracker,
max_cost_per_session: float = 1.00,
max_cost_per_step: float = 0.25,
alert_callback=None):
self.cost_tracker = cost_tracker
self.max_cost_per_session = max_cost_per_session
self.max_cost_per_step = max_cost_per_step
self.alert_callback = alert_callback # async function(session_id, message)
self.blocked_sessions = set()
async def check_step(self, session_id: str, step_cost: float):
"""Check if a single step exceeds limits."""
if session_id in self.blocked_sessions:
raise CostLimitExceeded(f"Session {session_id} is blocked")
if step_cost > self.max_cost_per_step:
if self.alert_callback:
await self.alert_callback(
session_id,
f"Step cost ${step_cost:.4f} exceeds max ${self.max_cost_per_step:.2f}"
)
self.blocked_sessions.add(session_id)
raise CostLimitExceeded(
f"Step cost ${step_cost:.4f} exceeds maximum ${self.max_cost_per_step:.2f}"
)
session_total = self.cost_tracker.get_session_total(session_id)
if session_total > self.max_cost_per_session:
if self.alert_callback:
await self.alert_callback(
session_id,
f"Session total ${session_total:.4f} exceeds max ${self.max_cost_per_session:.2f}"
)
self.blocked_sessions.add(session_id)
raise CostLimitExceeded(
f"Session total ${session_total:.4f} exceeds maximum ${self.max_cost_per_session:.2f}"
)
def is_session_healthy(self, session_id: str) -> bool:
return session_id not in self.blocked_sessions
class CostLimitExceeded(Exception):
passCore Pillar 3: Error Tracking
Categories of Agent Errors
AI agent errors differ from traditional exceptions. You need to track several distinct failure categories:
- Model-level errors ā API timeouts, rate limiting (429), server errors (5xx), content filter refusals
- Tool execution errors ā external API failures, invalid tool arguments, timeout on long-running operations
- Agent logic errors ā infinite loops, exceeding maximum steps, dead-end reasoning paths
- Guardrail violations ā output that fails content moderation, PII detection triggers, or compliance checks
- Partial failures ā the agent succeeded but skipped a critical step or produced degraded output
Structured Error Collector
The following error tracking system captures all failure categories with enough context for root-cause analysis:
# error_tracker.py
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional
import time
import json
class ErrorSeverity(Enum):
LOW = "low" # Degraded but acceptable
MEDIUM = "medium" # Actionable, needs investigation
HIGH = "high" # User-facing failure
CRITICAL = "critical" # System-wide impact
class ErrorCategory(Enum):
MODEL_API = "model_api"
MODEL_REFUSAL = "model_refusal"
TOOL_FAILURE = "tool_failure"
AGENT_LOOP = "agent_loop"
GUARDRAIL_VIOLATION = "guardrail_violation"
PARTIAL_FAILURE = "partial_failure"
TIMEOUT = "timeout"
@dataclass
class AgentError:
trace_id: str
session_id: str
step: int
category: ErrorCategory
severity: ErrorSeverity
message: str
context: dict
timestamp: float
class ErrorTracker:
"""Collects and aggregates agent errors for analysis."""
def __init__(self, max_buffer_size: int = 1000):
self.errors: List[AgentError] = []
self.max_buffer_size = max_buffer_size
self.error_counts_by_category: Dict[str, int] = {}
self.error_counts_by_severity: Dict[str, int] = {}
def record_error(self, trace_id: str, session_id: str, step: int,
category: ErrorCategory, severity: ErrorSeverity,
message: str, context: dict = None) -> AgentError:
"""Record an agent error with full context."""
error = AgentError(
trace_id=trace_id,
session_id=session_id,
step=step,
category=category,
severity=severity,
message=message,
context=context or {},
timestamp=time.time()
)
self.errors.append(error)
# Maintain rolling counts
cat_key = category.value
self.error_counts_by_category[cat_key] = \
self.error_counts_by_category.get(cat_key, 0) + 1
sev_key = severity.value
self.error_counts_by_severity[sev_key] = \
self.error_counts_by_severity.get(sev_key, 0) + 1
# Prune old errors if buffer exceeds limit
if len(self.errors) > self.max_buffer_size:
self.errors = self.errors[-self.max_buffer_size:]
return error
def get_recent_errors(self, minutes: int = 15) -> List[AgentError]:
"""Get errors from the last N minutes."""
cutoff = time.time() - (minutes * 60)
return [e for e in self.errors if e.timestamp >= cutoff]
def get_errors_by_session(self, session_id: str) -> List[AgentError]:
"""Retrieve all errors for a specific session."""
return [e for e in self.errors if e.session_id == session_id]
def get_high_severity_errors(self) -> List[AgentError]:
"""Return errors that need immediate attention."""
return [e for e in self.errors
if e.severity in (ErrorSeverity.HIGH, ErrorSeverity.CRITICAL)]
def generate_error_report(self) -> dict:
"""Produce a structured error report for dashboards."""
recent = self.get_recent_errors(minutes=60)
return {
"total_errors": len(self.errors),
"errors_last_hour": len(recent),
"by_category": self.error_counts_by_category.copy(),
"by_severity": self.error_counts_by_severity.copy(),
"high_severity_count": len(self.get_high_severity_errors()),
"most_recent": [
{
"trace_id": e.trace_id,
"session_id": e.session_id,
"step": e.step,
"category": e.category.value,
"severity": e.severity.value,
"message": e.message,
"timestamp": e.timestamp
}
for e in self.errors[-10:] # last 10 errors
]
}
# Usage integrated with agent loop
error_tracker = ErrorTracker()
# Example: recording a tool failure
try:
# Simulated tool call that fails
raise Exception("API returned 503 Service Unavailable")
except Exception as exc:
error_tracker.record_error(
trace_id="trace-xyz",
session_id="user-42-session-abc",
step=2,
category=ErrorCategory.TOOL_FAILURE,
severity=ErrorSeverity.HIGH,
message=f"Tool 'fetch_report' failed: {str(exc)}",
context={
"tool_name": "fetch_report",
"tool_input": {"quarter": "Q3"},
"retry_count": 0,
"http_status": 503
}
)
# Example: recording a model refusal
error_tracker.record_error(
trace_id="trace-xyz",
session_id="user-42-session-abc",
step=3,
category=ErrorCategory.MODEL_REFUSAL,
severity=ErrorSeverity.MEDIUM,
message="Model refused to process due to content filter",
context={
"model": "gpt-4o",
"refusal_reason": "content_policy_violation",
"prompt_snippet": "The user requested analysis of restricted document..."
}
)
report = error_tracker.generate_error_report()
print(json.dumps(report, indent=2))Error Recovery Patterns
Tracking errors is half the battle ā the other half is implementing graceful recovery. Common patterns include:
- Exponential backoff with jitter for transient API failures
- Fallback model routing ā if GPT-4o times out, retry with GPT-4o-mini
- Tool degradation ā if a primary tool fails, use a cached result or a simpler alternative
- Circuit breaking ā after N consecutive failures, stop the agent and return a graceful error to the user
Here's a recovery decorator that logs failures and implements exponential backoff:
# recovery_patterns.py
import asyncio
import random
from functools import wraps
from error_tracker import ErrorTracker, ErrorCategory, ErrorSeverity
def with_llm_recovery(max_retries: int = 3, base_delay_ms: int = 1000,
fallback_model: str = None, error_tracker: ErrorTracker = None):
"""Decorator for LLM calls with exponential backoff and optional fallback."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_error = e
if error_tracker:
error_tracker.record_error(
trace_id=kwargs.get("trace_id", "unknown"),
session_id=kwargs.get("session_id", "unknown"),
step=kwargs.get("step", 0),
category=ErrorCategory.MODEL_API,
severity=ErrorSeverity.MEDIUM if attempt < max_retries
else ErrorSeverity.HIGH,
message=f"LLM call failed (attempt {attempt+1}/{max_retries+1}): {str(e)}",
context={"attempt": attempt, "max_retries": max_retries}
)
if attempt < max_retries:
delay = base_delay_ms * (2 ** attempt) + random.uniform(0, 1000)
await asyncio.sleep(delay / 1000)
elif fallback_model:
kwargs["model"] = fallback_model
# One final attempt with fallback model
try:
return await func(*args, **kwargs)
except Exception as fallback_error:
raise fallback_error
raise last_error
return wrapper
return decorator
# Usage
@with_llm_recovery(max_retries=2, fallback_model="gpt-4o-mini",
error_tracker=error_tracker)
async def call_llm(model: str, prompt: str, trace_id: str,
session_id: str, step: int):
# Your actual LLM API call here
passBringing It All Together: The Unified Agent Monitor
The three pillars ā logging, cost tracking, and error tracking ā work best as a unified system. Below is a facade class that orchestrates all three, providing a single interface for instrumenting any agent loop:
# agent_monitor_unified.py
from agent_monitor import AgentLogger
from cost_tracker import CostTracker
from error_tracker import ErrorTracker, ErrorCategory, ErrorSeverity
from typing import Optional
import time
class UnifiedAgentMonitor:
"""Single interface combining logging, cost, and error tracking."""
def __init__(self):
self.logger = AgentLogger()
self.cost_tracker = CostTracker()
self.error_tracker = ErrorTracker()
self.current_trace_id: Optional[str] = None
def start_session(self, user_input: str, user_id: str,
session_id: str = None, feature: str = "default") -> str:
"""Begin a new monitored agent session."""
trace_id = self.logger.start_invocation(user_input, session_id)
self.current_trace_id = trace_id
self.cost_tracker.start_session(
session_id or trace_id, user_id, feature
)
return trace_id
def record_step(self, step: int, model: str,
prompt: str, response: str,
prompt_tokens: int, completion_tokens: int,
latency_ms: float):
"""Record an LLM call with full observability."""
trace_id = self.current_trace_id
session_id = trace_id # Using trace_id as session_id for simplicity
# Logging
self.logger.log_llm_call(
trace_id, step, model, prompt, response,
prompt_tokens, completion_tokens, latency_ms
)
# Cost tracking
step_cost = self.cost_tracker.record_llm_cost(
session_id, step, model, prompt_tokens, completion_tokens
)
return step_cost
def record_tool(self, step: int, tool_name: str,
tool_input: dict, tool_output: dict,
success: bool, duration_ms: float,
error_message: str = None):
"""Record a tool execution with error tracking if needed."""
trace_id = self.current_trace_id
self.logger.log_tool_call(
trace_id, step, tool_name, tool_input, tool_output,
success, duration_ms, error_message
)
if not success:
self.error_tracker.record_error(
trace_id=trace_id,
session_id=trace_id,
step=step,
category=ErrorCategory.TOOL_FAILURE,
severity=ErrorSeverity.HIGH,
message=error_message or f"Tool '{tool_name}' failed",
context={
"tool_name": tool_name,
"tool_input": tool_input,
"tool_output": tool_output,
"duration_ms": duration_ms
}
)
def finish_session(self, final_output: str) -> dict:
"""Complete the session and return a comprehensive summary."""
trace_id = self.current_trace_id
session_id = trace_id
cost_breakdown = self.cost_tracker.get_session_breakdown(session_id)
error_report = self.error_tracker.generate_error_report()
total_latency = time.time() - self.cost_tracker.session_metadata.get(
session_id, {}
).get("start_time", time.time())
self.logger.finish_invocation(
trace_id, final_output,
total_cost=cost_breakdown["total"],
total_latency_ms=total_latency * 1000
)
return {
"trace_id": trace_id,
"final_output": final_output,
"cost_breakdown": cost_breakdown,
"error_summary": {
"total_errors": error_report["total_errors"],
"high_severity": error_report["high_severity_count"],
"categories": error_report["by_category"]
},
"total_latency_ms": total_latency * 1000
}
# Full agent loop example
async def run_monitored_agent(user_query: str, user_id: str):
monitor = UnifiedAgentMonitor()
trace_id = monitor.start_session(user_query, user_id, feature="qa_agent")
# Step 1: Initial reasoning
step1_cost = monitor.record_step(
step=1, model="gpt-4o",
prompt=f"You are a helpful assistant. User query: {user_query}",
response="I need to search the knowledge base for this answer.",
prompt_tokens=120, completion_tokens=30,
latency_ms=1200
)
# Step