Ai Ai Observability

AI Observability Agent

Purpose

Design and implement end-to-end observability for AI systems: LLM tracing, token/cost tracking, latency monitoring, feedback collection, guardrail effectiveness, drift detection, and production alerting. Covers the full pipeline from agent instrumentation to collector to storage to dashboard.

Decision Trees

Scale Decision Tree

How many LLM calls per day?
├──  1,000,000
    ├── OpenTelemetry collector with tail-based sampling processor
    ├── Metrics aggregation at edge (statsd/stadshaper sidecar)
    ├── Storage: cloud-native (S3/GCS for traces, time-series DB for metrics)
    └── Dashboard: Grafana with hierarchical views

Budget Decision Tree

Monthly observability budget?
├──  $10,000/month
    ├── Enterprise Datadog, Honeycomb, or self-managed OpenTelemetry stack
    ├── Dedicated observability engineer
    └── Custom retention policies per data class

Stack Decision Tree

Existing infrastructure?
├── Kubernetes-native
│   ├── OpenTelemetry Collector (daemonset) → Tempo (traces) + Cortex (metrics) + Loki (logs)
│   └── Grafana dashboards with LLM-specific panels
├── Serverless / Lambda
│   ├── OTel Lambda layers → collector (Lambda extension) → backends
│   └── Managed: Datadog Serverless APM, Lumigo
├── Monolithic / Single service
│   ├── LangFuse or Helicone (minimal setup)
│   └── Native SDK tracing → built-in dashboard
└── Multi-service / Microservices
    ├── OpenTelemetry with context propagation (W3C traceparent)
    ├── Central collector per service mesh
    └── Trace correlation across service boundaries

Core Patterns

Pattern 1: Tracing (OpenTelemetry for LLM Calls)

Use OpenTelemetry semantic conventions for generative AI (semconv gen_ai). Every LLM call produces a span with:

• gen_ai.system: "openai", "anthropic", "google", "azure"
• gen_ai.request.model: model identifier
• gen_ai.request.max_tokens, gen_ai.request.temperature
• gen_ai.response.model: resolved model name
• gen_ai.usage.prompt_tokens, gen_ai.usage.completion_tokens, gen_ai.usage.total_tokens
• gen_ai.response.finish_reason: "stop", "length", "content_filter"

Trace every chain step, tool invocation, retriever query, and guardrail check as child spans. Root span carries userid, sessionid, and application version.

Pattern 2: Metrics (Three Pillars)

Collect three metric categories with consistent label schemas:

Latency metrics:

llm_request_duration_ms{model, operation, provider}  # Histogram
llm_time_to_first_token_ms{model, provider}           # Histogram
llm_inter_token_latency_ms{model}                     # Gauge
llm_queue_wait_ms{queue_name}                         # Histogram

Token & cost metrics:

llm_token_usage_total{model, direction}               # Counter (direction: input/output)
llm_cost_usd_total{model, provider, team}             # Counter
llm_cost_per_query{model}                             # Gauge

Quality & safety metrics:

llm_feedback_score{model, category}                   # Gauge (0-1)
llm_guardrail_violations{guardrail_type, severity}    # Counter
llm_hallucination_rate{model}                         # Gauge
llm_toxicity_score{model, category}                   # Gauge
llm_refusal_rate{model}                               # Gauge

Pattern 3: Logging (Structured, Searchable)

Emit structured JSON logs for every LLM interaction:

{
  "timestamp": "2026-05-31T10:00:00Z",
  "level": "info",
  "event": "llm_completion",
  "trace_id": "abc123",
  "span_id": "def456",
  "model": "gpt-4o",
  "system_fingerprint": "fp_abc",
  "prompt_truncated": true,
  "prompt_hash": "sha256:...",
  "response_hash": "sha256:...",
  "input_tokens": 450,
  "output_tokens": 120,
  "total_tokens": 570,
  "cost_usd": 0.00375,
  "latency_ms": 1234,
  "finish_reason": "stop",
  "user_id": "usr_abc",
  "session_id": "sess_xyz",
  "application": "chat-app",
  "environment": "prod"
}

Ship logs via OTel log exporter or stdin redirect to collector. Index by trace_id for full-context debugging. Hash prompt contents for deduplication without storing raw PII.

Pattern 4: Alerting (Multi-Window, Multi-Burn-Rate)

Configure alerts with burn-rate approach using two windows (short and long):

# Burn-rate alert for latency SLO
alert: HighLatencyBurnRate
expr: (
  rate(llm_request_duration_ms_count{model="gpt-4o"}[1m])  # short window
  /
  rate(llm_request_duration_ms_count{model="gpt-4o"}[1h])  # long window
) - 1 > 0.1
for: 2m
labels:
  severity: critical
  slo: latency_p95_2s
annotations:
  summary: "Latency burn rate {{ $value | humanizePercentage }} for gpt-4o"

Alert tiers:

• P0: Safety incident, data leakage, full outage → 5min response
• P1: Quality degradation >10%, cost spike >3x → 15min response
• P2: Latency regression, minor availability dip → 1h response
• P3: Budget warning, trend alert → next business day

Pattern 5: Drift Detection

Monitor embedding drift between baseline and production windows:

• Compute mean embedding vector over baseline period (e.g., last 7 days)
• Compute mean over sliding window (e.g., last 1 hour)
• Track Euclidean / cosine distance between the two
• Alert when distance exceeds 3 standard deviations from historical norm
• Segment drift by model, prompt template, user segment

Observability Pipeline Architecture

┌─────────────────────────────────────────────────────────┐
│                    APPLICATION LAYER                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌─────────┐ │
│  │ LLM Call │  │  Chain   │  │   Tool   │  │Guardrail│ │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬────┘ │
│       │             │             │             │       │
│  ┌────┴─────────────┴─────────────┴─────────────┴────┐ │
│  │           OpenTelemetry SDK (Traces + Metrics)     │ │
│  │           + Structured Logger (Logs)               │ │
│  └───────────────────────┬───────────────────────────┘ │
└──────────────────────────┼─────────────────────────────┘
                           │
┌──────────────────────────┼─────────────────────────────┐
│                  COLLECTOR LAYER                        │
│  ┌───────────────────────┴───────────────────────────┐ │
│  │           OpenTelemetry Collector                  │ │
│  │                                                    │ │
│  │  Receivers:    Processors:        Exporters:       │ │
│  │  ┌─────────┐  ┌──────────────┐  ┌──────────────┐ │ │
│  │  │ OTLP    │  │ batch        │  │ Tempo/Cortex │ │ │
│  │  │ Prometheus│ │ tail_sampling│  │ Loki         │ │ │
│  │  │ Filelog │  │ attributes   │  │ S3/GCS       │ │ │
│  │  └─────────┘  │ transform    │  │ Datadog      │ │ │
│  │               │ filter       │  └──────────────┘ │ │
│  │               │ k8s_atttributes                 │ │ │
│  │               └──────────────┘                   │ │
│  └───────────────────────────────────────────────────┘ │
└──────────────────────────┼─────────────────────────────┘
                           │
┌──────────────────────────┼─────────────────────────────┐
│                   STORAGE LAYER                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │   Traces     │  │   Metrics    │  │    Logs      │ │
│  │ (Tempo/     │  │ (Cortex/     │  │ (Loki/       │ │
│  │  Jaeger/    │  │  Mimir/      │  │  Elastic-    │ │
│  │  Datadog)   │  │  Datadog)    │  │  search)     │ │
│  └──────┬──────┘  └──────┬──────┘  └──────┬───────┘ │
│         │               │               │          │
│  ┌──────┴───────────────┴───────────────┴───────┐  │
│  │         Object Store (S3/GCS) for            │  │
│  │         long-term trace archival             │  │
│  └──────────────────────────────────────────────┘  │
└──────────────────────────┼─────────────────────────────┘
                           │
┌──────────────────────────┼─────────────────────────────┐
│                  DASHBOARD LAYER                        │
│  ┌───────────────────────────────────────────────────┐ │
│  │  Grafana / Datadog / LangFuse Dashboard            │ │
│  │                                                    │ │
│  │  Row 1: Overview (RPS, active users, error rate)   │ │
│  │  Row 2: Latency (P50/P95/P99 heatmap by model)     │ │
│  │  Row 3: Cost (daily stack by model/category/user)  │ │
│  │  Row 4: Quality (feedback score, guardrails, drift) │ │
│  │  Row 5: Alerts (firing, silenced, acknowledged)     │ │
│  └───────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

Pipeline Configuration Reference

OpenTelemetry Collector config (tail-based sampling):

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 1s
    send_batch_size: 1024
  tail_sampling:
    decision_wait: 30s
    num_traces: 50000
    policies:
      - name: errors-policy
        type: status_code
        config: { status_code: ERROR }
      - name: slow-policy
        type: latency
        config: { threshold_ms: 5000 }
      - name: probabilistic-policy
        type: probabilistic
        config: { sampling_percentage: 10 }

exporters:
  otlp/tempo:
    endpoint: tempo:4317
    tls: { insecure: true }
  prometheus:
    endpoint: 0.0.0.0:8889
  loki:
    endpoint: http://loki:3100/loki/api/v1/push

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [tail_sampling, batch]
      exporters: [otlp/tempo]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [loki]

Code Examples

Instrumenting LLM Calls with OpenTelemetry (Generic)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import openai

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

def tracked_llm_call(messages: list, model: str = "gpt-4o", user_id: str = None, session_id: str = None) -> str:
    with tracer.start_as_current_span("llm.completion") as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.request.max_tokens", 4096)
        span.set_attribute("gen_ai.request.temperature", 0.7)
        span.set_attribute("user_id", user_id or "anonymous")
        span.set_attribute("session_id", session_id or "unknown")

        response = openai.chat.completions.create(model=model, messages=messages)

        span.set_attribute("gen_ai.response.model", response.model)
        span.set_attribute("gen_ai.usage.prompt_tokens", response.usage.prompt_tokens)
        span.set_attribute("gen_ai.usage.completion_tokens", response.usage.completion_tokens)
        span.set_attribute("gen_ai.usage.total_tokens", response.usage.total_tokens)
        span.set_attribute("gen_ai.response.finish_reason", response.choices[0].finish_reason)

        if response.usage.completion_tokens > 0:
            span.set_attribute("gen_ai.response.latency_ms", response.response_ms)

        return response.choices[0].message.content

LangFuse Integration (Observe Decorator)

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse()

@observe(name="chat_agent", as_type="agent")
def chat_agent(query: str, user_id: str, session_id: str):
    langfuse_context.update_current_trace(
        user_id=user_id,
        session_id=session_id,
        metadata={"environment": "production", "app_version": "2.1.0"},
    )

    with langfuse_context.span(name="retrieve_context", type="retrieval") as span:
        docs = vector_store.similarity_search(query, k=5)
        span.update(input=query, output=[d.page_content for d in docs])
        langfuse_context.update_current_observation(
            usage={"input": len(query.split()), "output": sum(len(d.page_content.split()) for d in docs)}
        )

    with langfuse_context.generation(
        name="llm_response",
        model="gpt-4o",
        model_parameters={"temperature": 0.7, "max_tokens": 1000},
    ) as gen:
        response = openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": build_system_prompt(docs)},
                {"role": "user", "content": query},
            ],
        )
        content = response.choices[0].message.content
        gen.update(
            input=query,
            output=content,
            usage={
                "input": response.usage.prompt_tokens,
                "output": response.usage.completion_tokens,
                "unit": "TOKENS",
            },
        )

    langfuse_context.score_current_trace(
        name="response_quality",
        value=compute_relevance(query, content),
    )

    return content

LangSmith Integration (Decorator + Metadata)

from langsmith import traceable
from langsmith.run_helpers import get_current_run
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-app"

@traceable(run_type="chain", project_name="my-app")
def qa_chain(question: str, user_id: str) -> dict:
    run = get_current_run()
    run.add_metadata({
        "user_id": user_id,
        "session_id": f"session_{user_id}",
        "app_version": "2.1.0",
        "feature": "qa",
    })

    docs = retriever.get_relevant_documents(question)
    response = llm.invoke(format_prompt(question, docs))

    run.add_outputs({
        "output": response,
        "source_documents": [d.metadata["source"] for d in docs],
        "tokens": {"input": count_tokens(question), "output": count_tokens(response)},
    })

    return {"answer": response, "sources": [d.metadata["source"] for d in docs]}

Custom Metrics Pipeline (Prometheus Client)

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

llm_requests = Counter("llm_requests_total", "Total LLM requests", ["model", "status"])
llm_latency = Histogram(
    "llm_latency_ms", "LLM latency in ms", ["model"],
    buckets=[50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000],
)
llm_tokens = Counter("llm_tokens_total", "Total tokens used", ["model", "direction"])
llm_cost = Counter("llm_cost_usd", "Total cost in USD", ["model", "team"])
active_users = Gauge("llm_active_users", "Current active users", ["tier"])

def instrumented_call(model: str, team: str, user_id: str):
    start = time.time()
    try:
        response = call_llm(model)
        latency = (time.time() - start) * 1000
        llm_requests.labels(model=model, status="success").inc()
        llm_latency.labels(model=model).observe(latency)
        llm_tokens.labels(model=model, direction="input").inc(response.usage.prompt_tokens)
        llm_tokens.labels(model=model, direction="output").inc(response.usage.completion_tokens)
        cost = calculate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens)
        llm_cost.labels(model=model, team=team).inc(cost)
        return response
    except Exception as e:
        llm_requests.labels(model=model, status="error").inc()
        raise

start_http_server(8000)  # Prometheus scrape endpoint

Cost-Per-Trace Analytics

class CostPerTraceAnalyzer:
    def __init__(self):
        self.pricing = {
            "gpt-4o": {"input": 0.0025, "output": 0.01},
            "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
            "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
            "claude-3-haiku": {"input": 0.00025, "output":

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [j4flmao](https://github.com/j4flmao)
- **Source:** [j4flmao/agent-skills](https://github.com/j4flmao/agent-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.