Monitoring & Observability

Your AI agent is live. Traffic is flowing. But how do you know if it's actually helping customers? Traditional monitoring tells you the API is up and responses are fast. It doesn't tell you if the agent just recommended a product that's been recalled, or if it's consistently misunderstanding shipping questions.

AI agents need a different kind of observability.

The Observability Gap

Building the Observability Stack

┌─────────────────────────────────────────────────────────────┐
│                    Dashboards & Alerts                       │
│         (Grafana, Datadog, Custom Dashboards)               │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                    Analytics Layer                           │
│    Quality Metrics │ Cost Analysis │ User Behavior          │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│              LLM Observability Platform                      │
│         (LangSmith, Langfuse, Weights & Biases)             │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                 Structured Logging                           │
│     Traces │ Spans │ Events │ Metrics                       │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                    AI Agent Service                          │
└─────────────────────────────────────────────────────────────┘

Comprehensive Logging

Every agent interaction should capture structured data:

from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Dict, Any
import json
import uuid

@dataclass
class AgentSpan:
    """A single step in agent execution"""
    span_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    parent_span_id: Optional[str] = None
    name: str = ""
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    attributes: Dict[str, Any] = field(default_factory=dict)
    events: List[Dict] = field(default_factory=list)
    status: str = "running"

@dataclass
class AgentTrace:
    """Complete trace of an agent interaction"""
    trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    conversation_id: str = ""
    user_id: str = ""
    spans: List[AgentSpan] = field(default_factory=list)

    # Input/Output
    input_query: str = ""
    output_response: str = ""

    # Timing
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None

    # LLM Details
    model_used: str = ""
    total_tokens: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    estimated_cost: float = 0.0

    # Quality Signals
    tools_called: List[str] = field(default_factory=list)
    retrieval_results: int = 0
    confidence_score: Optional[float] = None
    cache_hit: bool = False

    # Outcome
    user_feedback: Optional[int] = None
    was_escalated: bool = False
    follow_up_count: int = 0


class AgentLogger:
    """Structured logging for AI agents"""

    def __init__(self, service_name: str):
        self.service = service_name
        self.current_traces: Dict[str, AgentTrace] = {}

    def start_trace(
        self,
        conversation_id: str,
        user_id: str,
        query: str
    ) -> AgentTrace:
        trace = AgentTrace(
            conversation_id=conversation_id,
            user_id=user_id,
            input_query=query
        )
        self.current_traces[trace.trace_id] = trace
        return trace

    def start_span(
        self,
        trace_id: str,
        name: str,
        parent_span_id: Optional[str] = None,
        attributes: Dict = None
    ) -> AgentSpan:
        span = AgentSpan(
            name=name,
            parent_span_id=parent_span_id,
            attributes=attributes or {}
        )
        self.current_traces[trace_id].spans.append(span)
        return span

    def log_llm_call(
        self,
        trace_id: str,
        span_id: str,
        model: str,
        prompt: str,
        response: str,
        tokens: Dict[str, int],
        latency_ms: float
    ):
        """Log an LLM API call with full details"""
        trace = self.current_traces[trace_id]

        # Find and update span
        for span in trace.spans:
            if span.span_id == span_id:
                span.attributes.update({
                    "llm.model": model,
                    "llm.prompt_preview": prompt[:500],
                    "llm.response_preview": response[:500],
                    "llm.input_tokens": tokens["input"],
                    "llm.output_tokens": tokens["output"],
                    "llm.latency_ms": latency_ms
                })
                break

        # Update trace totals
        trace.input_tokens += tokens["input"]
        trace.output_tokens += tokens["output"]
        trace.total_tokens += tokens["input"] + tokens["output"]
        trace.model_used = model

    def log_tool_call(
        self,
        trace_id: str,
        tool_name: str,
        input_params: Dict,
        output: Any,
        latency_ms: float,
        success: bool
    ):
        """Log a tool/function call"""
        trace = self.current_traces[trace_id]
        trace.tools_called.append(tool_name)

        span = self.start_span(
            trace_id,
            f"tool:{tool_name}",
            attributes={
                "tool.name": tool_name,
                "tool.input": json.dumps(input_params),
                "tool.output_preview": str(output)[:200],
                "tool.latency_ms": latency_ms,
                "tool.success": success
            }
        )
        span.end_time = datetime.now()
        span.status = "success" if success else "error"

    def end_trace(
        self,
        trace_id: str,
        response: str,
        confidence: Optional[float] = None,
        escalated: bool = False
    ):
        """Complete a trace with final details"""
        trace = self.current_traces[trace_id]
        trace.end_time = datetime.now()
        trace.output_response = response
        trace.confidence_score = confidence
        trace.was_escalated = escalated
        trace.estimated_cost = self._calculate_cost(trace)

        # Export to observability platform
        self._export_trace(trace)

        del self.current_traces[trace_id]
        return trace

Integrating with LangSmith

LangSmith provides specialized LLM observability:

from langsmith import Client
from langsmith.run_trees import RunTree
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "ecommerce-agent-prod"

class LangSmithInstrumentation:
    def __init__(self):
        self.client = Client()

    async def traced_agent_call(
        self,
        agent,
        query: str,
        context: dict,
        conversation_id: str
    ):
        """Wrap agent call with LangSmith tracing"""

        # Create run tree for this interaction
        run_tree = RunTree(
            name="ecommerce_agent",
            run_type="chain",
            inputs={
                "query": query,
                "conversation_id": conversation_id,
                "user_context": {
                    k: v for k, v in context.items()
                    if k not in ["pii_fields"]  # Don't log PII
                }
            },
            extra={
                "metadata": {
                    "environment": "production",
                    "agent_version": "2.1.0"
                }
            }
        )

        try:
            # Execute with tracing
            response = await agent.handle(query, context)

            run_tree.end(
                outputs={"response": response},
                error=None
            )

            return response

        except Exception as e:
            run_tree.end(error=str(e))
            raise

        finally:
            # Post run for analysis
            run_tree.post()


# With LangChain, tracing is automatic
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor

# All calls automatically traced to LangSmith
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# View traces at smith.langchain.com

Key Metrics to Track

Performance Metrics

class AgentMetrics:
    """Essential metrics for production AI agents"""

    def __init__(self, metrics_client):
        self.metrics = metrics_client

    def record_request(self, trace: AgentTrace):
        # Latency
        latency_ms = (trace.end_time - trace.start_time).total_seconds() * 1000
        self.metrics.histogram(
            "agent.latency_ms",
            latency_ms,
            tags={"model": trace.model_used}
        )

        # Token usage
        self.metrics.increment(
            "agent.tokens.input",
            trace.input_tokens,
            tags={"model": trace.model_used}
        )
        self.metrics.increment(
            "agent.tokens.output",
            trace.output_tokens,
            tags={"model": trace.model_used}
        )

        # Cost
        self.metrics.increment(
            "agent.cost_usd",
            trace.estimated_cost,
            tags={"model": trace.model_used}
        )

        # Cache performance
        self.metrics.increment(
            "agent.cache_hit" if trace.cache_hit else "agent.cache_miss"
        )

        # Tool usage
        for tool in trace.tools_called:
            self.metrics.increment(
                "agent.tool_calls",
                tags={"tool": tool}
            )

    def record_quality(self, conversation_id: str, metrics: dict):
        # User satisfaction (if provided)
        if "rating" in metrics:
            self.metrics.gauge("agent.user_rating", metrics["rating"])

        # Escalation
        if metrics.get("escalated"):
            self.metrics.increment("agent.escalations")

        # Follow-up questions (proxy for incomplete answers)
        self.metrics.histogram(
            "agent.followup_count",
            metrics.get("followup_count", 0)
        )

Quality Signals Dashboard

Alerting Strategy

class AgentAlerts:
    """Define alerts for agent health"""

    ALERT_RULES = {
        # Immediate alerts (PagerDuty)
        "critical": [
            {
                "name": "high_error_rate",
                "condition": "error_rate > 5%",
                "window": "5m",
                "action": "page_oncall"
            },
            {
                "name": "latency_spike",
                "condition": "p99_latency > 30s",
                "window": "5m",
                "action": "page_oncall"
            },
        ],

        # Warning alerts (Slack)
        "warning": [
            {
                "name": "escalation_spike",
                "condition": "escalation_rate > 15%",
                "window": "1h",
                "action": "slack_alert"
            },
            {
                "name": "cost_anomaly",
                "condition": "hourly_cost > 2x_baseline",
                "window": "1h",
                "action": "slack_alert"
            },
            {
                "name": "low_satisfaction",
                "condition": "avg_rating < 3.5",
                "window": "4h",
                "action": "slack_alert"
            },
        ],

        # Info alerts (Daily digest)
        "info": [
            {
                "name": "cache_degradation",
                "condition": "cache_hit_rate < 20%",
                "window": "24h",
                "action": "daily_digest"
            },
            {
                "name": "new_query_patterns",
                "condition": "unclassified_queries > 10%",
                "window": "24h",
                "action": "daily_digest"
            },
        ]
    }

Debugging with Traces

When something goes wrong, traces enable root cause analysis:

class TraceAnalyzer:
    """Analyze traces to find issues"""

    async def find_failed_conversations(
        self,
        time_range: tuple,
        filters: dict = None
    ) -> List[AgentTrace]:
        """Find conversations that didn't go well"""

        query = {
            "time_range": time_range,
            "filters": {
                "$or": [
                    {"was_escalated": True},
                    {"user_feedback": {"$lt": 3}},
                    {"followup_count": {"$gt": 3}},
                ]
            }
        }

        if filters:
            query["filters"].update(filters)

        return await self.trace_store.query(query)

    async def analyze_failure_patterns(
        self,
        failed_traces: List[AgentTrace]
    ) -> Dict[str, Any]:
        """Find common patterns in failed conversations"""

        analysis = {
            "common_intents": Counter(),
            "tool_failures": Counter(),
            "model_distribution": Counter(),
            "avg_tokens": [],
            "sample_queries": []
        }

        for trace in failed_traces:
            # Classify the query intent
            intent = await self.classifier.classify(trace.input_query)
            analysis["common_intents"][intent] += 1

            # Check for tool failures
            for span in trace.spans:
                if span.name.startswith("tool:") and span.status == "error":
                    analysis["tool_failures"][span.name] += 1

            analysis["model_distribution"][trace.model_used] += 1
            analysis["avg_tokens"].append(trace.total_tokens)
            analysis["sample_queries"].append(trace.input_query[:100])

        return {
            "top_failing_intents": analysis["common_intents"].most_common(5),
            "tool_issues": analysis["tool_failures"].most_common(5),
            "models_involved": dict(analysis["model_distribution"]),
            "avg_token_usage": sum(analysis["avg_tokens"]) / len(analysis["avg_tokens"]),
            "sample_queries": analysis["sample_queries"][:10]
        }

Real-World: Intercom's Fin Observability

Intercom's AI agent "Fin" processes millions of support conversations. Their observability approach:

Key Takeaways

Next up: Agent Quality & Evaluation - How to systematically measure and improve agent performance.