agent-observability

Agent Observability & Monitoring Skill

When to Use

Implement this skill when your multi-agent system needs observability across agent spans, LLM calls, token costs, latency, error rates, and alerting. Covers the full stack: trace export → metrics collection → structured logging → dashboards → incident response.

OpenTelemetry Tracing

Span Creation & Context Propagation

from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.propagate import inject, extract
import json

resource = Resource.create({"service.name": "agent-orchestrator"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

class AgentTracer:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.tracer = trace.get_tracer(agent_name)

    async def trace_agent_run(self, task: dict) -> dict:
        with self.tracer.start_as_current_span(
            f"agent.{self.agent_name}.run",
            kind=SpanKind.SERVER,
            attributes={
                "agent.name": self.agent_name,
                "task.id": task.get("id"),
                "task.type": task.get("type", "unknown"),
            },
        ) as span:
            try:
                result = await self._execute(task)
                span.set_attribute("result.status", "success")
                span.set_attribute("result.length", len(str(result)))
                return result
            except Exception as e:
                span.set_status(Status(StatusCode.ERROR, str(e)))
                span.record_exception(e)
                raise

    async def _execute(self, task: dict) -> dict:
        with self.tracer.start_as_current_span(
            f"agent.{self.agent_name}.execute",
            kind=SpanKind.INTERNAL,
            attributes={"task.priority": task.get("priority", 0)},
        ) as span:
            spans = await self._llm_call(task["prompt"])
            return spans