# Ai Ai Observability

> >

- **Type:** Skill
- **Install:** `agentstack add skill-j4flmao-agent-skills-ai-observability`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [j4flmao](https://agentstack.voostack.com/s/j4flmao)
- **Installs:** 0
- **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [j4flmao](https://github.com/j4flmao)
- **Source:** https://github.com/j4flmao/agent-skills/tree/main/skills/ai/ai-observability

## Install

```sh
agentstack add skill-j4flmao-agent-skills-ai-observability
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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 user_id, session_id, 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:
```json
{
  "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):
```yaml
# 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):**
```yaml
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)
```python
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)
```python
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)
```python
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)
```python
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
```python
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.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** yes
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** yes
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-j4flmao-agent-skills-ai-observability
- Seller: https://agentstack.voostack.com/s/j4flmao
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.