AI Agent Observability — How to Monitor Agents in Production

# agent_tracing.py — OpenTelemetry tracing for AI agents
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
import time, uuid, json

# Setup tracer
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter("http://localhost:4317")))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("zoo-agent")

class TracedAgent:
    def __init__(self, agent_id, model="gpt-4o"):
        self.agent_id = agent_id
        self.model = model
        self.total_cost = 0
        self.call_count = 0

    async def execute(self, task: str) -> dict:
        trace_id = str(uuid.uuid4())
        with tracer.start_as_current_span(
            f"agent.{self.agent_id}.execute",
            attributes={"agent.id": self.agent_id, "trace.id": trace_id}
        ) as root_span:
            try:
                # Step 1: Planning
                plan = await self._trace_step("plan", task, root_span)

                # Step 2: Tool calls (parallel)
                results = await self._trace_parallel_tools(plan, root_span)

                # Step 3: Synthesis
                output = await self._trace_step("synthesize", results, root_span)

                root_span.set_attribute("agent.success", True)
                root_span.set_attribute("agent.total_cost", self.total_cost)
                return {"output": output, "trace_id": trace_id, "cost": self.total_cost}

            except Exception as e:
                root_span.set_attribute("agent.success", False)
                root_span.set_attribute("error.message", str(e))
                raise

    async def _trace_step(self, step_name, input_data, parent_span):
        with tracer.start_as_current_span(
            f"step.{step_name}",
            attributes={"agent.id": self.agent_id, "step.name": step_name}
        ) as span:
            start = time.time()
            # ... actual LLM call here ...
            latency = time.time() - start
            span.set_attribute("llm.latency_ms", latency * 1000)
            span.set_attribute("llm.model", self.model)
            return output

# cost_monitor.py — Real-time cost tracking for AI agents
from dataclasses import dataclass, field
from datetime import datetime, date
import asyncio

@dataclass
class AgentBudget:
    agent_id: str
    daily_limit: float = 5.00      # $5/day default
    per_task_limit: float = 0.50   # $0.50 per task max
    alert_threshold: float = 0.8    # Alert at 80% of budget
    _spend: dict = field(default_factory=dict)

    def record_spend(self, task_id: str, cost: float) -> dict:
        today = date.today().isoformat()
        self._spend.setdefault(today, {"total": 0.0, "tasks": {}})
        self._spend[today]["total"] += cost
        self._spend[today]["tasks"][task_id] = cost

        # Check per-task limit
        if cost > self.per_task_limit:
            return {"action": "alert", "reason": f"Task {task_id} cost ${cost:.3f} > ${self.per_task_limit} limit"}

        # Check daily threshold
        daily = self._spend[today]["total"]
        if daily > self.daily_limit * self.alert_threshold:
            return {"action": "warn", "reason": f"Daily spend ${daily:.2f} at {daily/self.daily_limit*100:.0f}% of budget"}
        if daily > self.daily_limit:
            return {"action": "block", "reason": f"Daily budget exhausted: ${daily:.2f}/${self.daily_limit}"}

        return {"action": "ok"}

# Usage in agent execution loop
budgets = {agent_id: AgentBudget(agent_id) for agent_id in AGENT_IDS}

async def execute_with_budget(agent_id, task, llm_call):
    budget = budgets[agent_id]
    result = await llm_call(task)
    cost = result["total_tokens"] * PRICING[result["model"]]

    status = budget.record_spend(task["id"], cost)
    if status["action"] == "block":
        raise BudgetExhaustedError(status["reason"])
    return result

# quality_scorer.py — Automated output quality checks
from enum import Enum

class QualityDimension(Enum):
    RELEVANCE = "relevance"      # Does it address the task?
    COMPLETENESS = "completeness"  # All required fields present?
    SAFETY = "safety"            # No PII leaks, no harmful content
    FORMAT = "format"            # Valid JSON, correct schema
    HALLUCINATION = "hallucination"  # Factual accuracy check

async def score_output(task: str, output: str, schema: dict = None) -> dict:
    scores = {}

    # 1. Format validation (cheap, synchronous)
    if schema:
        try:
            parsed = json.loads(output)
            validate(parsed, schema)
            scores[QualityDimension.FORMAT] = 1.0
        except:
            scores[QualityDimension.FORMAT] = 0.0

    # 2. Relevance check (LLM-as-judge, lightweight model)
    relevance_prompt = f"Rate relevance 0-1. Task: {task} Output: {output[:500]}"
    scores[QualityDimension.RELEVANCE] = await llm_score(relevance_prompt)

    # 3. Safety scan (regex + classifier)
    scores[QualityDimension.SAFETY] = await safety_check(output)

    # 4. Hallucination detection (RAG-based fact check)
    if has_claims(output):
        scores[QualityDimension.HALLUCINATION] = await fact_check(output)

    overall = sum(scores.values()) / len(scores)
    return {"scores": scores, "overall": overall, "pass": overall >= 0.7}