Monitoring Observability

Monitoring & Observability

Comprehensive patterns for infrastructure monitoring, LLM observability, and quality drift detection. Each category has individual rule files in rules/ loaded on-demand.

Quick Reference

Total: 12 rules across 4 categories

Quick Start

# Prometheus metrics with RED method
from prometheus_client import Counter, Histogram

http_requests = Counter('http_requests_total', 'Total requests', ['method', 'endpoint', 'status'])
http_duration = Histogram('http_request_duration_seconds', 'Request latency',
    buckets=[0.01, 0.05, 0.1, 0.5, 1, 2, 5])

# Langfuse v4 LLM tracing — semantic as_type + inline scoring
from langfuse import observe, get_client

@observe(as_type="generation", name="analyze_content")
async def analyze_content(content: str):
    get_client().update_current_trace(
        user_id="user_123", session_id="session_abc",
        tags=["production", "orchestkit"],
    )
    result = await llm.generate(content)
    get_client().score_current_span(name="response_quality", value=0.85)
    return result

# PSI drift detection
import numpy as np

psi_score = calculate_psi(baseline_scores, current_scores)
if psi_score >= 0.25:
    alert("Significant quality drift detected!")

Infrastructure Monitoring

Prometheus metrics, Grafana dashboards, and alerting for application health.

CC 2.1.161 — OTEL resource attributes as metric labels: OTEL_RESOURCE_ATTRIBUTES values are now attached as labels on metric datapoints, so usage metrics can be sliced by custom dimensions (team, repo, environment). Add label selectors to dashboards for multi-tenant / per-team cost and usage tracking.

LLM Observability

Langfuse-based tracing, cost tracking, and evaluation for LLM applications.

Drift Detection

Statistical and quality drift detection for production LLM systems.

Silent Failures

Detection and alerting for silent failures in LLM agents.

CC 2.1.169 — OTEL client-cert paths require trust: untrusted project settings can no longer set OTEL client-certificate paths without a trust confirmation. If your OTEL exporter uses client certs configured in project .claude/settings.json, expect a one-time trust prompt on first use in an untrusted project — telemetry silently not flowing after 2.1.169 is usually this gate, not the collector.

Key Decisions

Detailed Documentation

Related Skills

• defense-in-depth - Layer 8 observability as part of security architecture
• devops-deployment - Observability integration with CI/CD and Kubernetes
• resilience-patterns - Monitoring circuit breakers and failure scenarios
• llm-evaluation - Evaluation patterns that integrate with Langfuse scoring
• caching - Caching strategies that reduce costs tracked by Langfuse

Langfuse Setup Checklist

Complete guide for setting up Langfuse observability in your application, based on OrchestKit's production implementation.

Prerequisites

• [ ] Python 3.10+ or Node.js 18+ application
• [ ] LLM integration (OpenAI, Anthropic, Google, etc.)
• [ ] PostgreSQL database (for self-hosted Langfuse)
• [ ] Docker and docker-compose (recommended for self-hosting)

Phase 1: Langfuse Server Setup

Option A: Langfuse Cloud (Fastest)

• [ ] Sign up at cloud.langfuse.com
• [ ] Create new project
• [ ] Copy LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY
• [ ] Copy LANGFUSE_HOST (usually https://cloud.langfuse.com)

Option B: Self-Hosted (Recommended for Production)

Langfuse v3 requires ClickHouse (analytics), Redis (queuing), MinIO (blob storage), and Postgres.

• [ ] Create docker-compose.yml for Langfuse:

  services:
    langfuse-web:
      image: langfuse/langfuse:3
      ports:
        - "3000:3000"
      environment:
        DATABASE_URL: postgresql://langfuse:CHANGE_ME_strong_password@postgres:5432/langfuse  # CHANGE ME
        CLICKHOUSE_URL: http://clickhouse:8123
        REDIS_URL: redis://redis:6379
        LANGFUSE_S3_UPLOAD_BUCKET: langfuse
        LANGFUSE_S3_ENDPOINT: http://minio:9000
        LANGFUSE_S3_ACCESS_KEY_ID: minio  # CHANGE ME for production
        LANGFUSE_S3_SECRET_ACCESS_KEY: miniosecret  # CHANGE ME for production
        NEXTAUTH_SECRET: your-secret-key-here  # Generate: openssl rand -base64 32
        NEXTAUTH_URL: http://localhost:3000
        SALT: your-salt-here  # Generate: openssl rand -base64 32
      depends_on:
        - postgres
        - clickhouse
        - redis
        - minio

    langfuse-worker:
      image: langfuse/langfuse-worker:3
      environment:
        DATABASE_URL: postgresql://langfuse:CHANGE_ME_strong_password@postgres:5432/langfuse  # CHANGE ME
        CLICKHOUSE_URL: http://clickhouse:8123
        REDIS_URL: redis://redis:6379
        LANGFUSE_S3_UPLOAD_BUCKET: langfuse
        LANGFUSE_S3_ENDPOINT: http://minio:9000
        LANGFUSE_S3_ACCESS_KEY_ID: minio  # CHANGE ME for production
        LANGFUSE_S3_SECRET_ACCESS_KEY: miniosecret  # CHANGE ME for production
      depends_on:
        - postgres
        - clickhouse
        - redis
        - minio

    postgres:
      image: postgres:15
      environment:
        POSTGRES_USER: langfuse
        POSTGRES_PASSWORD: password
        POSTGRES_DB: langfuse
      volumes:
        - langfuse-postgres:/var/lib/postgresql/data

    clickhouse:
      image: clickhouse/clickhouse-server:24
      environment:
        CLICKHOUSE_DB: langfuse
        CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
      volumes:
        - langfuse-clickhouse:/var/lib/clickhouse

    redis:
      image: redis:7-alpine
      volumes:
        - langfuse-redis:/data

    minio:
      image: minio/minio
      command: server /data --console-address ":9001"
      environment:
        MINIO_ROOT_USER: minio
        MINIO_ROOT_PASSWORD: miniosecret
      volumes:
        - langfuse-minio:/data

  volumes:
    langfuse-postgres:
    langfuse-clickhouse:
    langfuse-redis:
    langfuse-minio:

• [ ] Start Langfuse: docker-compose up -d
• [ ] Visit http://localhost:3000 and create admin account
• [ ] Create project in UI
• [ ] Copy API keys from Settings → API Keys

Phase 2: SDK Installation

Python (FastAPI/Flask/Django)

• [ ] Install SDK: pip install "langfuse>=4.0.0"
• [ ] Add to requirements.txt: langfuse>=4.0.0

Node.js (Express/Next.js)

• [ ] Install SDK: npm install @langfuse/core @langfuse/otel
• [ ] Add to package.json: "@langfuse/core": "^5.0.0" and "@langfuse/otel": "^5.0.0"

Phase 3: Configuration

Environment Variables

• [ ] Add to .env:

  LANGFUSE_PUBLIC_KEY=pk-lf-...
  LANGFUSE_SECRET_KEY=sk-lf-...
  LANGFUSE_HOST=http://localhost:3000  # or https://cloud.langfuse.com

• [ ] Add to .env.example (without values):

  LANGFUSE_PUBLIC_KEY=
  LANGFUSE_SECRET_KEY=
  LANGFUSE_HOST=

• [ ] Add to .gitignore: .env

Application Config

Python (backend/app/core/config.py):

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    LANGFUSE_PUBLIC_KEY: str
    LANGFUSE_SECRET_KEY: str
    LANGFUSE_HOST: str = "https://cloud.langfuse.com"

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

• [ ] Create settings class with Langfuse fields
• [ ] Validate environment variables on startup
• [ ] Add type hints for all config fields

Phase 4: Client Initialization

Python Client

File: backend/app/shared/services/langfuse/client.py

from langfuse import Langfuse
from app.core.config import settings

langfuse_client = Langfuse(
    public_key=settings.LANGFUSE_PUBLIC_KEY,
    secret_key=settings.LANGFUSE_SECRET_KEY,
    host=settings.LANGFUSE_HOST,
    debug=False,  # Set to True in development
    enabled=True  # Set to False to disable tracing
)

• [ ] Create dedicated client module
• [ ] Use singleton pattern for client instance
• [ ] Add debug mode for development
• [ ] Add enabled flag for testing/CI

Node.js Client

File: src/lib/langfuse.ts

import { Langfuse } from '@langfuse/core';

export const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: process.env.LANGFUSE_HOST || 'https://cloud.langfuse.com',
  debug: process.env.NODE_ENV === 'development',
  enabled: process.env.NODE_ENV !== 'test'
});

• [ ] Create dedicated client module
• [ ] Add TypeScript types
• [ ] Disable in test environment
• [ ] Enable debug mode in development

Phase 5: Decorator-Based Tracing

Python @observe Decorator

Example: backend/app/services/analysis.py

from langfuse import observe, get_client

@observe(name="analyze_content")
async def analyze_content(url: str, content: str) -> AnalysisResult:
    """Analyze content with automatic Langfuse tracing."""

    # Set trace-level metadata
    get_client().update_current_trace(
        name="content_analysis",
        session_id=f"analysis_{analysis_id}",
        user_id="system",
        metadata={
            "url": url,
            "content_length": len(content)
        },
        tags=["production", "v1"]
    )

    # Nested function - creates child span automatically
    @observe(name="fetch_metadata")
    async def fetch_metadata():
        # ... work ...
        pass

    # All nested calls create child spans
    metadata = await fetch_metadata()
    embedding = await generate_embedding(content)  # Also @observe decorated

    return AnalysisResult(metadata=metadata)

• [ ] Add @observe to all async functions that call LLMs
• [ ] Set meaningful span names
• [ ] Add session_id for multi-step workflows
• [ ] Add user_id for user-facing features
• [ ] Tag traces by environment (production/staging)

v4-Specific Setup

• [ ] Add as_type to @observe() decorators to classify span types ("span", "generation", "retriever", "chain")
• [ ] Add should_export_span filter to exclude noisy spans from export
• [ ] Migrate manual scoring to score_current_span() for simpler in-context scoring

Phase 6: LLM Call Instrumentation

Anthropic Claude

from langfuse import observe, get_client
from anthropic import AsyncAnthropic

anthropic_client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)

@observe(name="llm_call")
async def call_claude(prompt: str, model: str = "claude-sonnet-4-6") -> str:
    """Call Claude with cost tracking."""

    # Log input
    get_client().update_current_observation(
        input=prompt[:2000],  # Truncate large prompts
        model=model
    )

    # Call LLM
    response = await anthropic_client.messages.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=4096
    )

    # Extract tokens
    input_tokens = response.usage.input_tokens
    output_tokens = response.usage.output_tokens

    # Calculate cost (Claude Sonnet 4.6: $3/MTok input, $15/MTok output)
    cost_usd = (input_tokens / 1_000_000) * 3.00 + (output_tokens / 1_000_000) * 15.00

    # Log output and usage
    get_client().update_current_observation(
        output=response.content[0].text[:2000],
        usage={
            "input": input_tokens,
            "output": output_tokens,
            "unit": "TOKENS"
        },
        metadata={"cost_usd": cost_usd}
    )

    return response.content[0].text

• [ ] Wrap all LLM calls with @observe
• [ ] Log input/output (truncated)
• [ ] Track token usage
• [ ] Calculate and log costs
• [ ] Add model name to metadata

OpenAI

@observe(name="llm_call")
async def call_openai(prompt: str, model: str = "gpt-5.5") -> str:
    """Call OpenAI with cost tracking."""

    get_client().update_current_observation(
        input=prompt[:2000],
        model=model
    )

    response = await openai_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )

    # OpenAI pricing (gpt-5.5: $2.50/MTok input, $10/MTok output)
    input_tokens = response.usage.prompt_tokens
    output_tokens = response.usage.completion_tokens
    cost_usd = (input_tokens / 1_000_000) * 2.50 + (output_tokens / 1_000_000) * 10.00

    get_client().update_current_observation(
        output=response.choices[0].message.content[:2000],
        usage={
            "input": input_tokens,
            "output": output_tokens,
            "unit": "TOKENS"
        },
        metadata={"cost_usd": cost_usd}
    )

    return response.choices[0].message.content

• [ ] Add pricing for your models
• [ ] Update pricing when model costs change
• [ ] Log model name for each call

Phase 7: Quality Scoring

Add Evaluation Scores

from langfuse import observe, get_client

@observe(name="evaluate_quality")
async def evaluate_response(query: str, response: str) -> dict:
    """Evaluate LLM response quality."""

    # Run evaluation (your logic here)
    scores = {
        "relevance": 0.85,
        "coherence": 0.92,
        "depth": 0.78
    }

    # Add scores to current trace
    lf = get_client()
    trace_id = lf.get_current_trace_id()
    for criterion, score in scores.items():
        lf.create_score(
            trace_id=trace_id,
            name=criterion,
            value=score,
            comment=f"Evaluated {criterion} of response"
        )

    # Add overall score
    overall = sum(scores.values()) / len(scores)
    lf.create_score(
        trace_id=trace_id,
        name="overall_quality",
        value=overall,
        comment="Average of all criteria"
    )

    return scores

• [ ] Add quality scoring for all LLM outputs
• [ ] Use consistent criterion names
• [ ] Track scores over time
• [ ] Add comments explaining scores

Phase 8: Testing & Validation

Test Trace Creation

import pytest
from app.shared.services.langfuse.client import langfuse_client

@pytest.mark.asyncio
async def test_langfuse_trace_creation():
    """Verify Langfuse traces are created."""

    trace = langfuse_client.start_observation(
        name="test_trace",
        as_type="span",
        metadata={"test": True}
    )

    generation = trace.start_observation(
        name="test_generation",
        as_type="generation",
        model="claude-sonnet-4-6",
        input="Test prompt",
        output="Test response",
        usage_details={"input": 10, "output": 5}
    )
    generation.end()
    trace.end()

    # Flush to ensure data is sent
    langfuse_client.flush()

    assert trace.id is not None
    assert generation.id is not None

• [ ] Add integration tests for tracing
• [ ] Test trace creation
• [ ] Test score logging
• [ ] Verify data appears in UI

Verify in Langfuse UI

• [ ] Visit Langfuse UI
• [ ] Check Traces tab for test traces
• [ ] Verify metadata appears correctly
• [ ] Check Scores tab for quality metrics
• [ ] Verify cost calculations are accurate

Phase 9: Production Monitoring

Create Dashboards

• [ ] Cost Dashboard - Track spending by model, user, time
• [ ] Quality Dashboard - Monitor quality scores over time
• [ ] Performance Dashboard - Track latency by operation
• [ ] Error Dashboard - Failed traces, error rates

Set Up Alerts (via Langfuse UI or SQL)

-- Alert: Daily cost exceeds $100
SELECT
    DATE(timestamp) as date,
    SUM(calculated_total_cost) as daily_cost
FROM traces
WHERE timestamp > NOW() - INTERVAL '1 day'
GROUP BY DATE(timestamp)
HAVING SUM(calculated_total_cost) > 100;

• [ ] Daily cost threshold alerts
• [ ] Quality score degradation alerts
• [ ] High latency alerts
• [ ] Error rate alerts

Weekly Review Process

• [ ] Review top 10 most expensive traces
• [ ] Analyze quality score trends
• [ ] Identify optimization opportunities
• [ ] Update prompt versions based on scores

Phase 10: Advanced Features

Prompt Management

• [ ] Create prompts in Langfuse UI
• [ ] Version prompts using labels (production, staging, custom)
• [ ] Use get_client().get_prompt() in code with fallback= for resilience
• [ ] A/B test prompt versions
• [ ] Promote winning prompts to production label

Dataset Evaluation

• [ ] Create evaluation datasets in UI
• [ ] Run automated evaluations
• [ ] Track accuracy over time
• [ ] Compare model versions

Troubleshooting

Traces Not Appearing

• [ ] Check API keys are correct
• [ ] Verify LANGFUSE_HOST matches server
• [ ] Check enabled=True in client
• [ ] Call langfuse_client.flush() in tests
• [ ] Check network connectivity to Langfuse server

High Latency

• [ ] Enable async mode: flush_at=20 (batch sends)
• [ ] Reduce metadata size (truncate large strings)
• [ ] Use background thread for flushing

Missing Costs

• [ ] Verify usage data is logged: {"input": X, "output": Y, "unit": "TOKENS"}
• [ ] Check model pricing in Langfuse UI (Settings → Models)
• [ ] Add custom pricing if model not in database

References

• Langfuse Documentation
• Python SDK Guide
• Self-Hosting Guide
• Cost Tracking
• Template: ../scripts/observe-decorator.py

Monitoring Implementation Checklist

Complete guide for implementing production-grade monitoring, based on OrchestKit's real setup.

Prerequisites

• [ ] Application deployed (dev/staging/production)
• [ ] Docker or Kubernetes for monitoring stack
• [ ] Basic understanding of Prometheus, Grafana, Loki

Phase 1: Structured Logging

Python (structlog)

Install dependencies:

pip install structlog python-json-logger

• [ ] Install structlog and dependencies
• [ ] Add to requirements.txt

Configure structlog:

File: backend/app/core/logging.py

import logging
import structlog
from structlog.processors import JSONRenderer, TimeStamper, add_log_level

def configure_logging():
    """Configure structured logging with JSON output."""

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,  # Merge correlation IDs
            add_log_level,
            TimeStamper(fmt="iso"),
            JSONRenderer()
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True
    )

def get_logger(name: str):
    """Get a structured logger instance."""
    return structlog.get_logger(name)

• [ ] Create logging configuration module
• [ ] Configure JSON output (not plain text)
• [ ] Set appropriate log level (INFO for production)
• [ ] Add timestamp processor
• [ ] Enable context variable merging

Add correlation ID middleware:

import structlog
import uuid_utils  # pip install uuid-utils (UUID v7 for Python < 3.14)
from fastapi import Request

@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
    """Add correlation ID to all logs."""

    # Get or generate correlation ID (UUID v7 for time-ordering in traces)
    correlation_id = request.headers.get("X-Correlation-ID") or str(uuid_utils.uuid7())

    # Bind to logger context
    structlog.contextvars.bind_contextvars(
        correlation_id=correlation_id,
        method=request.method,
        path=request.url.path
    )

    # Process request
    response = await call_next(request)

    # Add to response headers
    response.headers["X-Correlation-ID"] = correlation_id

    # Clear context
    structlog.contextvars.clear_contextvars()

    return response

• [ ] Add correlation ID middleware
• [ ] Generate UUID if not provided
• [ ] Bind correlation_id to all logs in request
• [ ] Return correlation_id in response headers
• [ ] Clear context after request

Node.js (winston)

Install dependencies:

npm install winston express-winston uuid

• [ ] Install winston and dependencies
• [ ] Add to package.json

Configure winston:

File: src/lib/logger.ts

import winston from 'winston';

export const logger = winston.createLogger({
  level: process.env.LOG_LEVEL || 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.Console()
  ]
});

export const getLogger = (name: string) => {
  return logger.child({ logger: name });
};

• [ ] Create logger configuration
• [ ] Use JSON format
• [ ] Add timestamp to all logs
• [ ] Support child loggers with context

Phase 2: Metrics Collection

Python (prometheus-client)

Install:

pip install prometheus-client

• [ ] Install prometheus-client
• [ ] Add to requirements.txt

Create metrics module:

File: backend/app/core/metrics.py

from prometheus_client import Counter, Histogram, Gauge

# HTTP request metrics
http_requests_total = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

http_request_duration_seconds = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10]
)

# Database metrics
db_query_duration_seconds = Histogram(
    'db_query_duration_seconds',
    'Database query latency',
    ['query_type'],
    buckets=[0.001, 0.01, 0.05, 0.1, 0.5, 1]
)

db_connections_active = Gauge(
    'db_connections_active',
    'Number of active database connections'
)

# LLM metrics
llm_tokens_used = Counter(
    'llm_tokens_used_total',
    'Total LLM tokens consumed',
    ['model', 'operation', 'token_type']
)

llm_cost_dollars = Counter(
    'llm_cost_dollars_total',
    'Total LLM cost in dollars',
    ['model', 'operation']
)

# Cache metrics
cache_operations = Counter(
    'cache_operations_total',
    'Cache operations',
    ['operation', 'result']  # result=hit|miss
)

• [ ] Define HTTP metrics (requests, latency)
• [ ] Define database metrics (query latency, connections)
• [ ] Define LLM metrics (tokens, cost)
• [ ] Define cache metrics (hits, misses)
• [ ] Use appropriate metric types (Counter, Histogram, Gauge)
• [ ] Choose meaningful bucket boundaries

Add metrics middleware:

from fastapi import Request
import time
from app.core.metrics import http_requests_total, http_request_duration_seconds

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    """Track HTTP request metrics."""

    start_time = time.time()

    # Process request
    response = await call_next(request)

    # Record metrics
    duration = time.time() - start_time

    http_requests_total.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()

    http_request_duration_seconds.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)

    return response

• [ ] Add metrics middleware
• [ ] Track request count
• [ ] Track request duration
• [ ] Label by method, endpoint, status

Expose metrics endpoint:

from fastapi import Response
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

@app.get("/metrics")
async def metrics():
    """Expose Prometheus metrics."""
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

• [ ] Add /metrics endpoint
• [ ] Return Prometheus format
• [ ] Secure endpoint (internal network only)

Node.js (prom-client)

Install:

npm install prom-client

• [ ] Install prom-client
• [ ] Add to package.json

Create metrics:

import { Counter, Histogram, register } from 'prom-client';

export const httpRequestsTotal = new Counter({
  name: 'http_requests_total',
  help: 'Total HTTP requests',
  labelNames: ['method', 'endpoint', 'status']
});

export const httpRequestDuration = new Histogram({
  name: 'http_request_duration_seconds',
  help: 'HTTP request latency',
  labelNames: ['method', 'endpoint'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10]
});

// Expose metrics endpoint
app.get('/metrics', async (req, res) => {
  res.set('Content-Type', register.contentType);
  res.end(await register.metrics());
});

• [ ] Define metrics with prom-client
• [ ] Add /metrics endpoint
• [ ] Use consistent label names

Phase 3: Prometheus Setup

Docker Compose

File: monitoring/docker-compose.yml

version: '3.8'

services:
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
      - ./prometheus/alerts:/etc/prometheus/alerts
      - prometheus-data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
      - grafana-data:/var/lib/grafana

  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    volumes:
      - ./loki/loki.yml:/etc/loki/local-config.yaml
      - loki-data:/loki

  promtail:
    image: grafana/promtail:latest
    volumes:
      - ./promtail/promtail.yml:/etc/promtail/config.yml
      - /var/log:/var/log
    command: -config.file=/etc/promtail/config.yml

volumes:
  prometheus-data:
  grafana-data:
  loki-data:

• [ ] Create docker-compose.yml
• [ ] Add Prometheus service
• [ ] Add Grafana service
• [ ] Add Loki + Promtail for logs
• [ ] Configure volumes for persistence
• [ ] Set retention periods

Prometheus Configuration

File: monitoring/prometheus/prometheus.yml

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'backend'
    static_configs:
      - targets: ['backend:8500']  # Your app's /metrics endpoint

  - job_name: 'frontend'
    static_configs:
      - targets: ['frontend:3000']

  - job_name: 'postgres'
    static_configs:
      - targets: ['postgres-exporter:9187']

  - job_name: 'redis'
    static_configs:
      - targets: ['redis-exporter:9121']

# Load alerting rules
rule_files:
  - '/etc/prometheus/alerts/*.yml'

# Alertmanager configuration
alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

• [ ] Create Prometheus config
• [ ] Add scrape targets for all services
• [ ] Configure scrape interval (15s recommended)
• [ ] Load alerting rules
• [ ] Configure Alertmanager

Phase 4: Alerting Rules

File: monitoring/prometheus/alerts/service.yml

groups:
- name: service-health
  interval: 30s
  rules:
  - alert: ServiceDown
    expr: up == 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "Service {{ $labels.job }} is down"

  - alert: HighErrorRate
    expr: |
      sum(rate(http_requests_total{status=~"5.."}[5m])) /
      sum(rate(http_requests_total[5m])) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Error rate above 5%"

  - alert: HighLatency
    expr: |
      histogram_quantile(0.95,
        rate(http_request_duration_seconds_bucket[5m])
      ) > 2
    for: 10m
    labels:
      severity: high
    annotations:
      summary: "p95 latency above 2s"

• [ ] Create alerting rules file
• [ ] Add service availability alerts
• [ ] Add error rate alerts
• [ ] Add latency alerts
• [ ] Set appropriate thresholds
• [ ] Add meaningful annotations

File: monitoring/prometheus/alerts/application.yml

groups:
- name: application-metrics
  interval: 1m
  rules:
  # Cache performance
  - alert: LowCacheHitRate
    expr: |
      sum(rate(cache_operations_total{result="hit"}[30m])) /
      sum(rate(cache_operations_total[30m])) < 0.70
    for: 1h
    labels:
      severity: medium
    annotations:
      summary: "Cache hit rate below 70%"

  # Database performance
  - alert: SlowQueries
    expr: |
      histogram_quantile(0.95,
        rate(db_query_duration_seconds_bucket[5m])
      ) > 0.5
    for: 10m
    labels:
      severity: high
    annotations:
      summary: "Database queries slow (p95 > 500ms)"

  # LLM cost
  - alert: HighDailyCost
    expr: sum(increase(llm_cost_dollars_total[24h])) > 50
    labels:
      severity: high
    annotations:
      summary: "Daily LLM cost exceeded $50"

• [ ] Add cache alerts
• [ ] Add database alerts
• [ ] Add LLM cost alerts
• [ ] Set severity levels correctly

Phase 5: Grafana Dashboards

Datasource Configuration

File: monitoring/grafana/datasources/prometheus.yml

apiVersion: 1
datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true

  - name: Loki
    type: loki
    access: proxy
    url: http://loki:3100

• [ ] Configure Prometheus datasource
• [ ] Configure Loki datasource
• [ ] Set Prometheus as default

Service Overview Dashboard

Create dashboard with:

1. Golden Signals Row:

• [ ] Latency (p50, p95, p99)
• [ ] Traffic (requests/second)
• [ ] Errors (error rate %)
• [ ] Saturation (CPU, memory)

2. Request Breakdown:

• [ ] Requests by endpoint
• [ ] Requests by status code
• [ ] Request rate over time

3. Dependencies:

• [ ] Database query latency
• [ ] Redis latency
• [ ] External API latency

4. Resources:

• [ ] CPU usage
• [ ] Memory usage
• [ ] Disk I/O
• [ ] Network I/O

Example Panel Queries

Latency Panel:

# p50 latency
histogram_quantile(0.5, rate(http_request_duration_seconds_bucket[5m]))

# p95 latency
histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))

# p99 latency
histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m]))

Traffic Panel:

sum(rate(http_requests_total[5m]))

Error Rate Panel:

sum(rate(http_requests_total{status=~"5.."}[5m])) /
sum(rate(http_requests_total[5m]))

• [ ] Add all key panels
• [ ] Use appropriate visualization types
• [ ] Add thresholds for red/yellow/green
• [ ] Set refresh interval (10s-30s)

Phase 6: Log Aggregation (Loki)

Loki Configuration

File: monitoring/loki/loki.yml

auth_enabled: false

server:
  http_listen_port: 3100

ingester:
  lifecycler:
    ring:
      kvstore:
        store: inmemory
      replication_factor: 1

schema_config:
  configs:
    - from: 2024-01-01
      store: boltdb-shipper
      object_store: filesystem
      schema: v11
      index:
        prefix: index_
        period: 24h

storage_config:
  boltdb_shipper:
    active_index_directory: /loki/index
    cache_location: /loki/cache
  filesystem:
    directory: /loki/chunks

limits_config:
  retention_period: 168h  # 7 days

• [ ] Create Loki config
• [ ] Set retention period
• [ ] Configure storage backend
• [ ] Set appropriate limits

Promtail Configuration

File: monitoring/promtail/promtail.yml

server:
  http_listen_port: 9080

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: docker
    docker_sd_configs:
      - host: unix:///var/run/docker.sock
    relabel_configs:
      - source_labels: ['__meta_docker_container_name']
        target_label: 'container'
      - source_labels: ['__meta_docker_container_log_stream']
        target_label: 'stream'

• [ ] Create Promtail config
• [ ] Configure log sources (Docker, files, etc.)
• [ ] Add labels for filtering
• [ ] Point to Loki endpoint

Phase 7: Testing & Validation

Test Metrics Collection

# Check metrics endpoint
curl http://localhost:8500/metrics

# Verify Prometheus scraping
curl http://localhost:9090/api/v1/targets

# Query metrics
curl 'http://localhost:9090/api/v1/query?query=http_requests_total'

• [ ] Verify /metrics endpoint works
• [ ] Check Prometheus targets are up
• [ ] Query metrics via API
• [ ] Verify labels are correct

Test Logging

# Check logs in Loki
curl -G 'http://localhost:3100/loki/api/v1/query' \
  --data-urlencode 'query={job="backend"}' \
  --data-urlencode 'limit=10'

• [ ] Verify logs appear in Loki
• [ ] Check JSON parsing works
• [ ] Verify labels are correct
• [ ] Test LogQL queries

Test Alerting

# Check alert rules loaded
curl http://localhost:9090/api/v1/rules

# Check active alerts
curl http://localhost:9090/api/v1/alerts

• [ ] Verify alert rules loaded
• [ ] Trigger test alert (cause error)
• [ ] Verify alert fires
• [ ] Check alert appears in Alertmanager

Phase 8: Production Deployment

Security Checklist

• [ ] Restrict /metrics endpoint to internal network
• [ ] Enable authentication for Grafana
• [ ] Use HTTPS for all dashboards
• [ ] Rotate Grafana admin password
• [ ] Set up RBAC for Grafana users
• [ ] Enable audit logging

Performance Checklist

• [ ] Set appropriate retention periods (Prometheus: 30d, Loki: 7d)
• [ ] Configure metric cardinality limits
• [ ] Enable query caching
• [ ] Set memory limits for Prometheus
• [ ] Monitor monitoring stack resource usage

Alerting Checklist

• [ ] Configure Alertmanager receivers (Slack, PagerDuty, email)
• [ ] Set up alert routing rules
• [ ] Add inhibition rules (suppress noisy alerts)
• [ ] Test alert delivery
• [ ] Create runbooks for all critical alerts
• [ ] Set up on-call schedule

Phase 9: Ongoing Maintenance

Daily Checks

• [ ] Review active alerts
• [ ] Check dashboard for anomalies
• [ ] Verify all scrape targets are up

Weekly Checks

• [ ] Review top 10 slowest endpoints
• [ ] Check error rate trends
• [ ] Review LLM cost trends
• [ ] Update dashboards as needed

Monthly Checks

• [ ] Review alert thresholds (tune for accuracy)
• [ ] Clean up unused metrics
• [ ] Update Prometheus/Grafana versions
• [ ] Review retention policies
• [ ] Audit dashboard access

References

• Template: ../scripts/structured-logging.ts
• Template: ../scripts/prometheus-metrics.ts
• Template: ../scripts/alerting-rules.yml
• Example: ../examples/orchestkit-monitoring-dashboard.md
• Prometheus Best Practices
• Grafana Documentation
• Loki Documentation

OrchestKit Langfuse Traces - Real Implementation

This document shows how OrchestKit uses Langfuse for end-to-end LLM observability across its 8-agent LangGraph workflow.

Overview

OrchestKit Analysis Pipeline:

• 8 specialized agents (Tech Comparator, Security Auditor, Implementation Planner, etc.)
• LangGraph supervisor pattern for orchestration
• Langfuse traces for cost tracking, performance monitoring, and debugging

Migration: LangSmith → Langfuse (December 2025)

• Self-hosted, open-source, free
• Better prompt management
• Native cost tracking
• Session-based grouping

Trace Architecture

Analysis Session Structure

content_analysis (session_id: analysis_550e8400)
├── fetch_content (0.3s)
│   └── metadata: {url, content_size_bytes: 45823}
├── generate_embedding (0.8s, $0.0002)
│   └── model: voyage-code-2
│   └── tokens: 11,456 input
└── supervisor_workflow (12.5s, $0.145)
    ├── supervisor_route_1 (0.1s)
    │   └── next_agent: tech_comparator
    ├── tech_comparator (2.1s, $0.018)
    │   ├── analyze_technologies (1.8s, $0.015)
    │   │   └── model: claude-sonnet-4-6
    │   │   └── tokens: 1,500 input, 1,000 output
    │   └── compress_findings (0.2s, $0.003)
    │       └── model: claude-sonnet-4-6
    │       └── tokens: 800 input, 400 output
    ├── supervisor_route_2 (0.1s)
    │   └── next_agent: security_auditor
    ├── security_auditor (2.3s, $0.021)
    │   └── ... (similar structure)
    ├── ... (6 more agents)
    └── quality_gate (1.2s, $0.012)
        ├── g_eval_completeness (0.4s, $0.004)
        ├── g_eval_accuracy (0.4s, $0.004)
        ├── g_eval_coherence (0.2s, $0.002)
        └── g_eval_depth (0.2s, $0.002)

Session Metrics:

• Total duration: 15.4s
• Total cost: $0.147
• Agents executed: 8
• Quality scores: completeness=0.85, accuracy=0.92, coherence=0.88, depth=0.78

Implementation Examples

1. Workflow-Level Tracing

File: backend/app/domains/analysis/workflows/content_analysis.py

from langfuse import observe, get_client
from app.shared.services.langfuse.client import langfuse_client

@observe(name="content_analysis_workflow")
async def run_content_analysis(analysis_id: str, url: str) -> AnalysisResult:
    """Analyze content with 8-agent supervisor workflow."""

    # Set session-level metadata
    get_client().update_current_trace(
        name="content_analysis",
        session_id=f"analysis_{analysis_id}",
        user_id="system",
        metadata={
            "analysis_id": analysis_id,
            "url": url,
            "workflow_type": "8-agent-supervisor",
            "version": "1.0.0"
        },
        tags=["production", "orchestkit", "langgraph"]
    )

    # Step 1: Fetch content (nested span)
    content = await fetch_content(url)  # @observe decorated

    # Step 2: Generate embedding (nested span with cost tracking)
    embedding = await generate_embedding(content)  # @observe decorated

    # Step 3: Run supervisor workflow (8 agents in parallel/sequential)
    findings = await run_supervisor_workflow(content)

    # Track total cost
    total_cost = sum(f.cost_usd for f in findings)
    get_client().update_current_observation(
        metadata={
            "total_agents": len(findings),
            "total_cost_usd": total_cost,
            "total_tokens": sum(f.token_count for f in findings)
        }
    )

    return AnalysisResult(findings=findings, total_cost=total_cost)

2. Agent-Level Tracing

File: backend/app/domains/analysis/workflows/nodes/agent_node.py

@observe(name="agent_execution")
async def execute_agent(
    agent_type: str,
    content: str,
    state: AnalysisState
) -> Finding:
    """Execute single agent with Langfuse tracing."""

    # Set agent-specific context
    get_client().update_current_observation(
        name=f"agent_{agent_type}",
        metadata={
            "agent_type": agent_type,
            "content_length": len(content),
            "correlation_id": state["correlation_id"]
        }
    )

    # Call LLM with automatic cost tracking
    response = await call_llm_with_tracing(
        agent_type=agent_type,
        content=content,
        state=state
    )

    # Score the response
    quality_scores = await score_agent_output(agent_type, response)

    # Add scores to trace
    for criterion, score in quality_scores.items():
        get_client().score(
            name=f"{agent_type}_{criterion}",
            value=score,
            data_type="NUMERIC"
        )

    return response

3. LLM Call Tracing with Cost Tracking

File: backend/app/shared/services/llm/anthropic_client.py

from langfuse import observe, get_client

@observe(name="llm_call")
async def call_anthropic(
    messages: list[dict],
    model: str = "claude-sonnet-4-6",
    **kwargs
) -> str:
    """Call Anthropic with automatic Langfuse cost tracking."""

    # Log input (truncated for large prompts)
    get_client().update_current_observation(
        input=str(messages)[:2000],
        model=model,
        metadata={
            "temperature": kwargs.get("temperature", 1.0),
            "max_tokens": kwargs.get("max_tokens", 4096)
        }
    )

    # Call Anthropic API
    response = await anthropic_client.messages.create(
        model=model,
        messages=messages,
        **kwargs
    )

    # Extract token usage
    input_tokens = response.usage.input_tokens
    output_tokens = response.usage.output_tokens

    # Cost calculation (Claude Sonnet 4.5 pricing)
    input_cost = (input_tokens / 1_000_000) * 3.00   # $3/MTok
    output_cost = (output_tokens / 1_000_000) * 15.00  # $15/MTok
    total_cost = input_cost + output_cost

    # Log output and costs to Langfuse
    get_client().update_current_observation(
        output=response.content[0].text[:2000],
        usage={
            "input": input_tokens,
            "output": output_tokens,
            "unit": "TOKENS"
        },
        metadata={
            "cost_usd": total_cost,
            "input_cost_usd": input_cost,
            "output_cost_usd": output_cost,
            "prompt_caching_enabled": kwargs.get("cache_control") is not None
        }
    )

    logger.info("llm_call_completed",
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost_usd=total_cost,
        cache_enabled=kwargs.get("cache_control") is not None
    )

    return response.content[0].text

4. Quality Gate Evaluation Tracing

File: backend/app/workflows/nodes/quality_gate_node.py

@observe(name="quality_gate")
async def quality_gate_node(state: AnalysisState) -> AnalysisState:
    """Evaluate aggregated findings with G-Eval scoring."""

    get_client().update_current_observation(
        metadata={
            "findings_count": len(state["findings"]),
            "analysis_id": state["analysis_id"]
        }
    )

    # Run G-Eval for 4 criteria in parallel
    criteria = ["completeness", "accuracy", "coherence", "depth"]

    scores = await asyncio.gather(*[
        evaluate_criterion(criterion, state["findings"])
        for criterion in criteria
    ])

    # Log individual criterion scores
    score_dict = {}
    for criterion, score in zip(criteria, scores):
        score_dict[criterion] = score
        get_client().score(
            name=f"quality_{criterion}",
            value=score,
            comment=f"G-Eval score for {criterion} criterion"
        )

    # Overall quality score (weighted average)
    overall_quality = (
        score_dict["completeness"] * 0.3 +
        score_dict["accuracy"] * 0.3 +
        score_dict["coherence"] * 0.2 +
        score_dict["depth"] * 0.2
    )

    get_client().score(
        name="quality_overall",
        value=overall_quality,
        comment="Weighted average of all criteria"
    )

    state["quality_scores"] = score_dict
    state["overall_quality"] = overall_quality

    return state

Real Metrics from Production

Cost Breakdown by Agent

Langfuse Query (Last 30 Days):

SELECT
    metadata->>'agent_type' as agent,
    COUNT(*) as executions,
    AVG(calculated_total_cost) as avg_cost,
    SUM(calculated_total_cost) as total_cost,
    AVG(input_tokens) as avg_input_tokens,
    AVG(output_tokens) as avg_output_tokens
FROM traces
WHERE metadata->>'agent_type' IS NOT NULL
    AND timestamp > NOW() - INTERVAL '30 days'
GROUP BY agent
ORDER BY total_cost DESC;

Results:

Insights:

• Security Auditor is most expensive (detailed vulnerability analysis)
• Quality Gate is cheapest (focused evaluation)
• Total monthly cost: $18.73 (145 analyses)
• Average per analysis: $0.129

Cache Hit Impact

Before Caching (Dec 2024):

• Monthly cost: $35,000 (projected annual: $420k)
• Average latency: 2.1s per LLM call

After Multi-Level Caching (Jan 2025):

• L1 (Prompt Cache): 90% hit rate → $31,500 saved (90% savings on cache hits)
• L2 (Semantic Cache): 75% hit rate on L1 misses → $2,625 saved (85% savings)
• Final monthly cost: $875
• Total savings: 97.5%
• Average latency: 5-10ms (semantic cache hit)

Langfuse Cache Analytics:

-- Cache hit rate by agent
SELECT
    metadata->>'agent_type' as agent,
    COUNT(*) FILTER (WHERE metadata->>'cache_hit' = 'true') as cache_hits,
    COUNT(*) as total_calls,
    ROUND(100.0 * COUNT(*) FILTER (WHERE metadata->>'cache_hit' = 'true') / COUNT(*), 2) as hit_rate_pct
FROM traces
WHERE metadata->>'agent_type' IS NOT NULL
GROUP BY agent
ORDER BY hit_rate_pct DESC;

Results:

Why security_auditor has low cache hit rate:

• Unique vulnerabilities per codebase
• Security context is highly specific
• Opportunity: Implement vulnerability pattern caching

Dashboard Queries

Top 10 Most Expensive Analyses

SELECT
    name,
    session_id,
    calculated_total_cost as cost_usd,
    timestamp,
    metadata->>'url' as analyzed_url,
    metadata->>'total_agents' as agents_executed
FROM traces
WHERE name = 'content_analysis_workflow'
ORDER BY calculated_total_cost DESC
LIMIT 10;

Quality Trend Over Time

SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'quality_completeness') as avg_completeness,
    AVG(value) FILTER (WHERE name = 'quality_accuracy') as avg_accuracy,
    AVG(value) FILTER (WHERE name = 'quality_coherence') as avg_coherence,
    AVG(value) FILTER (WHERE name = 'quality_depth') as avg_depth,
    AVG(value) FILTER (WHERE name = 'quality_overall') as avg_overall
FROM scores
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date;

Results (Last 30 Days):

Trend: Quality scores stable, depth improving (+4% since truncation fix)

Slow Trace Detection

-- Find traces slower than 2 standard deviations
WITH stats AS (
    SELECT
        AVG(latency_seconds) as mean,
        STDDEV(latency_seconds) as stddev
    FROM traces
    WHERE name = 'content_analysis_workflow'
)
SELECT
    t.session_id,
    t.latency_seconds,
    t.metadata->>'url' as url,
    t.timestamp
FROM traces t, stats s
WHERE t.name = 'content_analysis_workflow'
    AND t.latency_seconds > (s.mean + 2 * s.stddev)
ORDER BY t.latency_seconds DESC
LIMIT 20;

Best Practices from OrchestKit

1. Always use @observe decorator - Automatic parent-child span relationships 2. Set session_id for multi-step workflows - Group related traces together 3. Tag production vs staging - Filter by environment 4. Add agent_type to metadata - Enable cost/performance analysis by agent 5. Log truncated inputs/outputs - Keep traces small (2000 chars max) 6. Score all quality metrics - Enable quality trend monitoring 7. Track cache_hit in metadata - Measure caching effectiveness 8. Use correlation_id across services - Link to application logs

References

• Langfuse Self-Hosting Guide
• Python SDK Decorators
• Cost Tracking
• OrchestKit QUALITY_INITIATIVE_FIXES.md

OrchestKit Monitoring Dashboard - Real Implementation

This document shows OrchestKit's actual monitoring setup including metrics, dashboards, and alerting rules.

Overview

OrchestKit Monitoring Stack:

• Logs: Structlog (JSON) → Loki
• Metrics: Prometheus (RED + business metrics)
• Traces: Langfuse (LLM observability)
• Dashboards: Grafana
• Alerts: Prometheus Alertmanager → Slack

Key Metrics:

• LLM costs: $35k/year → $2-5k/year (95% reduction via caching)
• Retrieval pass rate: 91.6% (target: >90%)
• Quality gate pass rate: 85% (target: >80%)
• Hybrid search latency: 5ms (HNSW index)

Dashboard Structure

1. Service Overview Dashboard

Top Row - Golden Signals:

┌──────────────┬──────────────┬──────────────┬──────────────┐
│  Latency     │  Traffic     │  Errors      │  Saturation  │
│  p50: 245ms  │  12.5 req/s  │  0.3% (5xx)  │  CPU: 45%    │
│  p95: 680ms  │  (stable)    │  (good)      │  Mem: 62%    │
│  p99: 1.2s   │              │              │  (healthy)   │
└──────────────┴──────────────┴──────────────┴──────────────┘

Prometheus Queries:

# p95 latency
histogram_quantile(0.95,
  rate(http_request_duration_seconds_bucket[5m])
)

# Request rate
sum(rate(http_requests_total[5m]))

# Error rate (5xx)
sum(rate(http_requests_total{status=~"5.."}[5m])) /
sum(rate(http_requests_total[5m]))

# CPU saturation
avg(rate(process_cpu_seconds_total[5m])) * 100

2. LLM Observability Dashboard

Metrics Tracked:

• Cost per model (Claude, Gemini, Voyage)
• Token usage (input/output)
• Cache hit rates (L1: Prompt Cache, L2: Semantic Cache)
• LLM latency distribution

Cost Breakdown Panel:

# Total cost per day by model
sum(increase(llm_cost_dollars_total[1d])) by (model)

# Cost per operation
sum(increase(llm_cost_dollars_total[1h])) by (operation)

Example Results:

Cache Performance Panel:

# Cache hit rate
sum(rate(cache_operations_total{result="hit"}[5m])) /
sum(rate(cache_operations_total[5m]))

# Cost savings from cache (estimated)
sum(rate(cache_operations_total{result="hit"}[5m])) *
avg_over_time(llm_cost_dollars_total[1h])

Results:

3. Quality Metrics Dashboard

Panels: 1. Quality gate pass rate (target: >80%) 2. G-Eval scores by criterion (completeness, accuracy, coherence, depth) 3. Failed analyses count 4. Quality score distribution

Quality Gate Pass Rate:

# Pass rate over last 24h
sum(rate(quality_gate_passed_total[24h])) /
sum(rate(quality_gate_total[24h]))

G-Eval Scores (from Langfuse):

-- Track quality trends
SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'quality_completeness') as completeness,
    AVG(value) FILTER (WHERE name = 'quality_accuracy') as accuracy,
    AVG(value) FILTER (WHERE name = 'quality_coherence') as coherence,
    AVG(value) FILTER (WHERE name = 'quality_depth') as depth
FROM langfuse.scores
WHERE timestamp > NOW() - INTERVAL '7 days'
GROUP BY DATE(timestamp);

Example Results:

4. Database Performance Dashboard

Panels: 1. Query latency (p50/p95/p99) 2. Connection pool usage 3. Slow queries (>500ms) 4. Cache hit ratio

Query Latency:

# p95 query latency
histogram_quantile(0.95,
  rate(db_query_duration_seconds_bucket[5m])
) by (query_type)

Connection Pool:

# Active connections
db_connections_active

# Connection pool saturation
db_connections_active / db_connections_max

Real Metrics:

5. Retrieval Quality Dashboard

Metrics from Golden Dataset (98 analyses, 415 chunks):

Pass Rate:

# Retrieval pass rate (expected chunk in top-k)
sum(retrieval_pass_total) / sum(retrieval_total)

Results: 186/203 queries passed = 91.6% pass rate (target: >90%)

MRR by Difficulty:

-- Mean Reciprocal Rank by query difficulty
SELECT
    difficulty,
    COUNT(*) as queries,
    AVG(mrr) as avg_mrr
FROM retrieval_evaluation
GROUP BY difficulty;

Results:

Search Latency:

# Hybrid search latency (HNSW + BM25 RRF)
histogram_quantile(0.95,
  rate(search_duration_seconds_bucket[5m])
)

Results:

Comparison to IVFFlat:

• HNSW: 5ms
• IVFFlat: 85ms
• Speedup: 17x faster

Structured Logging Examples

Log Format

OrchestKit uses structlog with JSON output:

{
  "event": "supervisor_routing",
  "level": "info",
  "timestamp": "2025-01-21T10:30:45.123Z",
  "correlation_id": "abc-123-def",
  "analysis_id": "550e8400-e29b-41d4-a716-446655440000",
  "workflow_step": "supervisor",
  "agent": "tech_comparator",
  "remaining_agents": 7,
  "content_length": 45823,
  "logger": "app.workflows.supervisor"
}

Key Log Events

1. Analysis Started:

{
  "event": "analysis_started",
  "level": "info",
  "analysis_id": "550e8400-...",
  "url": "https://example.com/article",
  "content_type": "article"
}

2. Agent Execution:

{
  "event": "agent_execution_started",
  "level": "info",
  "agent_type": "security_auditor",
  "correlation_id": "abc-123-def",
  "analysis_id": "550e8400-..."
}

3. LLM Call:

{
  "event": "llm_call_completed",
  "level": "info",
  "model": "claude-sonnet-4-6",
  "operation": "security_audit",
  "input_tokens": 1800,
  "output_tokens": 1200,
  "cost_dollars": 0.021,
  "duration_seconds": 2.3,
  "cache_hit": false
}

4. Quality Gate:

{
  "event": "quality_gate_passed",
  "level": "info",
  "analysis_id": "550e8400-...",
  "quality_scores": {
    "completeness": 0.85,
    "accuracy": 0.92,
    "coherence": 0.88,
    "depth": 0.78
  },
  "overall_quality": 0.86,
  "passed": true
}

5. Error Logging:

{
  "event": "analysis_failed",
  "level": "error",
  "analysis_id": "550e8400-...",
  "error_type": "ValidationError",
  "error_message": "Quality gate failed: depth score too low",
  "quality_scores": {
    "depth": 0.45
  },
  "traceback": "...",
  "correlation_id": "abc-123-def"
}

Loki Queries (LogQL)

Find all errors in last hour:

{app="orchestkit-backend"} |= "ERROR" | json

Count errors by endpoint:

sum by (endpoint) (
  count_over_time({app="orchestkit-backend"} |= "ERROR" [5m])
)

Search for specific analysis:

{app="orchestkit-backend"}
| json
| analysis_id="550e8400-e29b-41d4-a716-446655440000"

p95 LLM latency from logs:

quantile_over_time(0.95,
  {app="orchestkit-backend"}
  | json
  | event="llm_call_completed"
  | unwrap duration_seconds [5m]
)

Alerting Rules

1. Service Availability

File: monitoring/prometheus/alerts/service.yml

groups:
- name: service-health
  interval: 30s
  rules:
  - alert: ServiceDown
    expr: up == 0
    for: 1m
    labels:
      severity: critical
      team: platform
    annotations:
      summary: "Service {{ $labels.job }} is down"
      description: "{{ $labels.instance }} has been down for 1 minute"
      runbook_url: "https://wiki.orchestkit.dev/runbooks/service-down"

  - alert: HighErrorRate
    expr: |
      sum(rate(http_requests_total{status=~"5.."}[5m])) /
      sum(rate(http_requests_total[5m])) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "High error rate detected"
      description: "Error rate is {{ $value | humanizePercentage }} (threshold: 5%)"

2. LLM Cost Alerts

File: monitoring/prometheus/alerts/llm-cost.yml

groups:
- name: llm-costs
  interval: 1h
  rules:
  - alert: DailyCostExceeded
    expr: |
      sum(increase(llm_cost_dollars_total[24h])) > 20
    labels:
      severity: high
      team: ai-ml
    annotations:
      summary: "Daily LLM cost exceeded $20"
      description: "Current daily cost: ${{ $value }}"

  - alert: UnexpectedCostSpike
    expr: |
      sum(rate(llm_cost_dollars_total[1h])) >
      sum(rate(llm_cost_dollars_total[1h] offset 24h)) * 2
    for: 2h
    labels:
      severity: high
    annotations:
      summary: "LLM cost spike detected"
      description: "Current hourly cost is 2x yesterday's average"

3. Quality Degradation

File: monitoring/prometheus/alerts/quality.yml

groups:
- name: quality-metrics
  interval: 5m
  rules:
  - alert: LowQualityGatePassRate
    expr: |
      sum(rate(quality_gate_passed_total[1h])) /
      sum(rate(quality_gate_total[1h])) < 0.80
    for: 30m
    labels:
      severity: high
      team: ml
    annotations:
      summary: "Quality gate pass rate below 80%"
      description: "Current pass rate: {{ $value | humanizePercentage }}"

  - alert: CacheHitRateDegraded
    expr: |
      sum(rate(cache_operations_total{result="hit"}[30m])) /
      sum(rate(cache_operations_total[30m])) < 0.70
    for: 1h
    labels:
      severity: medium
    annotations:
      summary: "Cache hit rate below 70%"
      description: "Cache performance degraded: {{ $value | humanizePercentage }}"

4. Database Performance

File: monitoring/prometheus/alerts/database.yml

groups:
- name: database-performance
  interval: 1m
  rules:
  - alert: SlowQueries
    expr: |
      histogram_quantile(0.95,
        rate(db_query_duration_seconds_bucket[5m])
      ) > 0.5
    for: 10m
    labels:
      severity: high
    annotations:
      summary: "p95 query latency exceeded 500ms"
      description: "Current p95: {{ $value }}s"

  - alert: ConnectionPoolExhausted
    expr: db_connections_active / db_connections_max > 0.9
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Database connection pool near capacity"
      description: "{{ $value | humanizePercentage }} of connections in use"

Alert Routing & Escalation

File: monitoring/alertmanager/config.yml

route:
  group_by: ['alertname', 'cluster', 'service']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: slack-default

  routes:
  # Critical alerts → Slack + PagerDuty
  - match:
      severity: critical
    receiver: pagerduty-critical
    continue: true  # Also send to Slack

  # High severity → Slack
  - match:
      severity: high
    receiver: slack-high

  # Medium/low → Slack (throttled)
  - match_re:
      severity: (medium|low)
    receiver: slack-low
    group_interval: 1h

receivers:
- name: slack-default
  slack_configs:
  - api_url: <slack_webhook_url>
    channel: '#alerts'
    title: '{{ .GroupLabels.alertname }}'
    text: '{{ range .Alerts }}{{ .Annotations.summary }}\n{{ end }}'

- name: pagerduty-critical
  pagerduty_configs:
  - service_key: <pagerduty_service_key>

Health Check Endpoints

1. Liveness Probe

Endpoint: GET /health Purpose: Is the application running?

@app.get("/health")
async def health_check():
    """Basic liveness check."""
    return {"status": "healthy"}

2. Readiness Probe

Endpoint: GET /ready Purpose: Is the application ready to serve traffic?

@app.get("/ready")
async def readiness_check():
    """Check if app can handle requests."""

    checks = {}

    # Database check
    try:
        await db.execute("SELECT 1")
        checks["database"] = {"status": "pass", "latency_ms": 5}
    except Exception as e:
        checks["database"] = {"status": "fail", "error": str(e)}

    # Redis check
    try:
        await redis.ping()
        checks["redis"] = {"status": "pass", "latency_ms": 2}
    except Exception as e:
        checks["redis"] = {"status": "fail", "error": str(e)}

    # Overall status
    all_healthy = all(c["status"] == "pass" for c in checks.values())
    status = "healthy" if all_healthy else "degraded"

    return {
        "status": status,
        "checks": checks,
        "version": "1.0.0",
        "uptime": int(time.time() - app.start_time)
    }

Response:

{
  "status": "healthy",
  "checks": {
    "database": {"status": "pass", "latency_ms": 5},
    "redis": {"status": "pass", "latency_ms": 2}
  },
  "version": "1.0.0",
  "uptime": 3600
}

References

• Template: ../scripts/structured-logging.ts
• Template: ../scripts/prometheus-metrics.ts
• Template: ../scripts/alerting-rules.yml
• OrchestKit Redis Connection
• OrchestKit Quality Initiative

Agent Observability

Trace multi-agent systems with Agent Graphs, new observation types, and rendered tool calls.

Agent Graphs (GA Nov 2025)

Agent Graphs provide visual execution flow for multi-agent systems. Langfuse automatically renders agent handoffs, tool calls, and decision points as an interactive graph.

Enabling Agent Graphs

from langfuse import observe, get_client

@observe(as_type="agent", name="supervisor")
async def supervisor_agent(query: str):
    """Supervisor agent that routes to specialists."""
    get_client().update_current_observation(
        metadata={"routing_strategy": "semantic"}
    )

    intent = await classify_intent(query)

    if intent == "code_review":
        return await code_review_agent(query)
    elif intent == "security_audit":
        return await security_audit_agent(query)
    else:
        return await general_agent(query)


@observe(as_type="agent", name="code_review")
async def code_review_agent(query: str):
    """Specialist agent for code review."""
    context = await retrieve_context(query)
    return await generate_review(context)


@observe(as_type="agent", name="security_audit")
async def security_audit_agent(query: str):
    """Specialist agent for security auditing."""
    vulnerabilities = await scan_code(query)
    return await generate_audit_report(vulnerabilities)

Result in Langfuse UI — Agent Graph View

supervisor (agent)
├── classify_intent (chain) → "code_review"
├── code_review (agent)
│   ├── retrieve_context (retriever) → 5 chunks
│   └── generate_review (generation) → $0.03
│       └── tool_call: analyze_diff → rendered inline
└── Total: 3.2s, $0.05

The Agent Graph View renders:

• Agent nodes with execution order
• Tool calls with inputs/outputs rendered inline
• Decision edges showing routing logic
• Timing breakdown per agent

New Observation Types

Langfuse v3 adds 7 observation types beyond generation and span:

Using Observation Types

from langfuse import observe, get_client

@observe(as_type="retriever", name="vector_search")
async def retrieve_context(query: str):
    """Retrieve relevant context from vector DB."""
    results = await vector_db.search(query, top_k=5)
    get_client().update_current_observation(
        metadata={
            "top_k": 5,
            "chunks_returned": len(results),
            "avg_similarity": sum(r.score for r in results) / len(results),
        }
    )
    return results


@observe(as_type="tool", name="web_search")
async def search_web(query: str):
    """Execute web search tool."""
    results = await tavily.search(query)
    get_client().update_current_observation(
        input=query,
        output=results[:3],  # Top 3 results
        metadata={"source": "tavily", "result_count": len(results)},
    )
    return results


@observe(as_type="guardrail", name="pii_filter")
async def check_pii(text: str):
    """Check for PII before sending to LLM."""
    has_pii = detect_pii(text)
    get_client().update_current_observation(
        input=text[:200],
        output={"has_pii": has_pii, "action": "blocked" if has_pii else "passed"},
        metadata={"check_type": "pii"},
    )
    if has_pii:
        raise PiiDetectedError("PII detected in input")
    return text


@observe(as_type="embedding", name="embed_query")
async def embed_query(text: str):
    """Generate embedding for query."""
    embedding = await embeddings.embed(text)
    get_client().update_current_observation(
        input=text,
        model="text-embedding-3-large",
        usage={"input_tokens": len(text.split())},
        metadata={"dimensions": len(embedding)},
    )
    return embedding


@observe(as_type="evaluator", name="relevance_judge")
async def evaluate_relevance(query: str, response: str):
    """Evaluate response relevance with LLM judge."""
    score = await llm_judge.evaluate(
        criteria="relevance",
        query=query,
        response=response,
    )
    get_client().update_current_observation(
        input={"query": query, "response": response[:500]},
        output={"score": score, "criteria": "relevance"},
    )
    # Each evaluator run creates its own inspectable trace
    return score

Rendered Tool Calls

In v3, tool calls within generations are rendered inline in the trace view:

@observe(as_type="agent")
async def coding_agent(task: str):
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        tools=[
            {"name": "read_file", "description": "Read a file", "input_schema": {...}},
            {"name": "write_file", "description": "Write a file", "input_schema": {...}},
        ],
        messages=[{"role": "user", "content": task}],
    )

    # Tool calls automatically rendered in Langfuse trace:
    # generation → tool_use: read_file(path="src/main.py")
    #           → tool_result: "def main():..."
    #           → tool_use: write_file(path="src/main.py", content="...")
    #           → tool_result: "OK"
    return response

Trace Log View

The Trace Log View provides a chronological log of all events within a trace, useful for debugging agent loops:

[00:00.000] agent:supervisor START
[00:00.050] chain:classify_intent START
[00:00.320] chain:classify_intent END → "security_audit"
[00:00.321] agent:security_audit START
[00:00.400] retriever:vector_search START
[00:00.650] retriever:vector_search END → 5 chunks
[00:00.651] guardrail:pii_filter START
[00:00.680] guardrail:pii_filter END → passed
[00:00.681] generation:analyze START
[00:02.100] generation:analyze END → $0.04
[00:02.101] evaluator:relevance_judge START
[00:02.500] evaluator:relevance_judge END → 0.92
[00:02.501] agent:security_audit END
[00:02.502] agent:supervisor END → Total: 2.5s, $0.05

Framework Integration Examples

LangGraph

from langfuse import observe

@observe(as_type="agent", name="langgraph_supervisor")
async def run_langgraph_workflow(query: str):
    """LangGraph workflow with automatic Langfuse tracing."""
    from langgraph.graph import StateGraph

    graph = StateGraph(AgentState)
    graph.add_node("researcher", researcher_node)
    graph.add_node("writer", writer_node)

    # Each node automatically creates nested observations
    # when decorated with @observe
    app = graph.compile()
    return await app.ainvoke({"query": query})

CrewAI

from langfuse import observe

@observe(as_type="agent", name="crewai_crew")
async def run_crew(task: str):
    """CrewAI crew with Langfuse tracing."""
    from crewai import Crew, Agent, Task

    researcher = Agent(name="researcher", role="Research analyst")
    writer = Agent(name="writer", role="Technical writer")

    crew = Crew(
        agents=[researcher, writer],
        tasks=[Agent(description=task, agent=researcher)],
    )
    # CrewAI has built-in Langfuse integration via callbacks
    return crew.kickoff()

OpenAI Agents SDK

from langfuse import observe

@observe(as_type="agent", name="openai_agent")
async def run_openai_agent(query: str):
    """OpenAI Agents SDK with Langfuse tracing."""
    from openai_agents import Agent, Runner

    agent = Agent(
        name="analyst",
        instructions="You are a code analyst.",
        model="gpt-5.5",
    )
    # OpenAI Agents SDK supports Langfuse via OTEL exporter
    result = await Runner.run(agent, query)
    return result.final_output

Best Practices

1. Use `type="agent"` for autonomous agents that make routing decisions 2. Use `type="tool"` for function calls to external services 3. Use `type="retriever"` for all RAG retrieval steps 4. Use `type="guardrail"` for safety checks (PII, toxicity, etc.) 5. Use `type="evaluator"` for quality judges — each creates an inspectable trace 6. Add metadata with routing decisions, chunk counts, similarity scores 7. Name observations descriptively — they appear in the Agent Graph

References

• Agent Graphs
• Observation Types
• Tool Calls

Alerting and Dashboards

Effective alerting strategies and dashboard design patterns.

Alert Severity Levels

Key Alerts

Alert Grouping

Group related alerts:

route:
  group_by: ['alertname', 'cluster', 'service']
  group_wait: 30s        # Wait 30s to collect similar alerts
  group_interval: 5m     # Send grouped alerts every 5m
  repeat_interval: 4h    # Re-send alert after 4h if still firing

  routes:
  - match:
      severity: critical
    receiver: pagerduty
    continue: true        # Continue to other routes

  - match:
      severity: warning
    receiver: slack

Inhibition Rules

Suppress noisy alerts when root cause is known:

inhibit_rules:
# If ServiceDown is firing, suppress HighErrorRate and HighLatency
- source_match:
    alertname: ServiceDown
  target_match_re:
    alertname: (HighErrorRate|HighLatency)
  equal: ['service']

# If DatabaseDown is firing, suppress all DB-related alerts
- source_match:
    alertname: DatabaseDown
  target_match_re:
    alertname: Database.*
  equal: ['cluster']

Escalation Policies

# Escalation: Slack -> PagerDuty after 15 min
routes:
- match:
    severity: critical
  receiver: slack
  continue: true
  routes:
  - match:
      severity: critical
    receiver: pagerduty
    group_wait: 15m  # Escalate to PagerDuty after 15 min

Runbook Links

Add runbook links to alert annotations:

groups:
- name: app-alerts
  rules:
  - alert: HighErrorRate
    expr: |
      sum(rate(http_requests_total{status=~"5.."}[5m])) /
      sum(rate(http_requests_total[5m])) > 0.05
    annotations:
      summary: "High error rate detected"
      description: "Error rate is {{ $value | humanizePercentage }}"
      runbook_url: "https://wiki.example.com/runbooks/high-error-rate"

Runbook should include: 1. What the alert means 2. Impact on users 3. Common causes 4. Investigation steps 5. Remediation steps 6. Escalation contacts

Dashboard Design Principles

Golden Signals Dashboard (top row)

+--------------+--------------+--------------+--------------+
|  Latency     |  Traffic     |  Errors      |  Saturation  |
|  (p50/p95)   |  (req/s)     |  (5xx rate)  |  (CPU/mem)   |
+--------------+--------------+--------------+--------------+

Service Dashboard Structure

1. Overview (single row) - Traffic, errors, latency, saturation 2. Request breakdown - By endpoint, method, status code 3. Dependencies - Database, Redis, external APIs 4. Resources - CPU, memory, disk, network 5. Business metrics - Registrations, purchases, etc.

RED Metrics for Dashboards

• Rate: rate(http_requests_total[5m])
• Errors: sum(rate(http_requests_total{status=~"5.."}[5m]))
• Duration: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))

USE Metrics for Resources

• Utilization: % of resource used
• Saturation: Queue depth, wait time
• Errors: Error count

SLO/SLI Dashboards

Service Level Indicators (SLIs):

# Availability SLI: % of successful requests
sum(rate(http_requests_total{status!~"5.."}[30d])) /
sum(rate(http_requests_total[30d]))

# Latency SLI: % of requests < 1s
sum(rate(http_request_duration_seconds_bucket{le="1"}[30d])) /
sum(rate(http_request_duration_seconds_count[30d]))

Service Level Objectives (SLOs):

• Availability: 99.9% (43 min downtime/month)
• Latency: 99% of requests < 1s

Error Budget:

• 99.9% SLO = 0.1% error budget
• If error budget consumed, freeze feature work and focus on reliability

Notification Channels

• PagerDuty - critical (on-call)
• Slack - warnings (team channel)
• Email - low priority (daily digest)

See scripts/alerting-rules.yml for complete examples.

Alerting Strategies

Effective alerting to minimize false positives.

Alerting Rules (Prometheus)

groups:
  - name: api_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High error rate detected"
          description: "Error rate is {{ $value }}% over last 5 minutes"

      - alert: HighLatency
        expr: histogram_quantile(0.95, http_request_duration_seconds_bucket) > 1
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "High API latency"

Alert Severity Levels

Best Practices

1. Alert on symptoms, not causes - "Users can't login" not "CPU high" 2. Actionable alerts only - every alert needs runbook 3. Reduce noise - use for: 5m to avoid flapping 4. Group related alerts - don't page for every instance 5. Test alert rules - amtool alert query

Notification Channels

• PagerDuty - critical (on-call)
• Slack - warnings (team channel)
• Email - low priority (daily digest)

See scripts/alerting-rules.yml for complete examples.

Annotation Queues

Human-review workflow in Langfuse: route traces to reviewers, collect scores, feed decisions into golden dataset curation.

What Are Annotation Queues?

Annotation queues let you route specific traces to human reviewers for scoring. Reviewers see the trace inputs, outputs, and existing automated scores, then add a human judgment score. The collected scores become ground truth for evaluator calibration and golden dataset inclusion decisions.

Typical uses:

• Spot-checking high-cost or high-stakes LLM outputs
• Reviewing traces flagged as low-quality by automated evaluators
• Building ground-truth labels for fine-tuning or RAG evaluation

Creating Queues via Langfuse UI

1. Navigate to Annotation Queues in the left sidebar 2. Click Create Queue 3. Configure:

• Name — e.g., quality-review, safety-check, golden-dataset-candidates
• Description — what reviewers should evaluate
• Score configs — which scoring dimensions to collect (e.g., accuracy, relevance, safety)

4. Share the queue URL with reviewers — no code access required

Score configs define what the reviewer sees and scores. Create them under Settings → Score Configs before creating the queue.

Adding Traces to a Queue Programmatically

Use get_client().create_annotation_queue_item() inside an @observe-decorated function to route a trace for human review:

from langfuse import observe, get_client

@observe(name="generate-response")
async def generate_and_flag(user_query: str, queue_id: str) -> str:
    """Generate a response and flag low-confidence outputs for human review."""
    response = await llm.generate(user_query)
    score = await auto_evaluate(response)

    # Flag for human review when automated confidence is low
    if score < 0.7:
        lf = get_client()
        trace_id = lf.get_current_trace_id()
        lf.create_annotation_queue_item(
            queue_id=queue_id,
            trace_id=trace_id,
        )

    return response

You can also add traces to a queue outside of an @observe context using a standalone client:

from langfuse import Langfuse

lf = Langfuse()

# Add a known trace ID to a review queue
lf.create_annotation_queue_item(
    queue_id="queue-abc123",
    trace_id="trace-xyz789",
)

Retrieve queue IDs programmatically via the Langfuse API or copy them from the UI URL.

Human-Review Workflow

Trace in Langfuse
      |
      v
[Automated score < threshold]
      |
      v
create_annotation_queue_item()  →  Queue
                                      |
                                      v
                              Reviewer opens queue URL
                                      |
                              Views: input / output / existing scores
                                      |
                              Adds human scores (accuracy, safety, etc.)
                                      |
                                      v
                          Scores stored on trace in Langfuse
                                      |
                                      v
                    [Optional] Trigger golden dataset inclusion

Reviewers access queues via the Langfuse UI — no SDK or code access required. The reviewer sees:

• The trace input and output
• Any automated scores already applied
• The score dimensions configured for the queue

After scoring, human scores appear on the trace alongside automated scores and are queryable via the Langfuse API.

Fetching Completed Annotations

Query finished annotation items to drive downstream automation (e.g., auto-include high-scored traces into the golden dataset):

import httpx
import base64
import os

LANGFUSE_HOST = os.environ["LANGFUSE_HOST"]
PUBLIC_KEY = os.environ["LANGFUSE_PUBLIC_KEY"]
SECRET_KEY = os.environ["LANGFUSE_SECRET_KEY"]

auth = base64.b64encode(f"{PUBLIC_KEY}:{SECRET_KEY}".encode()).decode()
headers = {"Authorization": f"Basic {auth}"}

async def fetch_completed_annotations(queue_id: str) -> list[dict]:
    """Fetch completed annotation queue items."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{LANGFUSE_HOST}/api/public/annotation-queues/{queue_id}/items",
            headers=headers,
            params={"status": "DONE"},
        )
        resp.raise_for_status()
        return resp.json()["data"]

async def promote_high_quality_to_dataset(queue_id: str, dataset_name: str):
    """Add human-approved traces to golden dataset."""
    from langfuse import Langfuse
    lf = Langfuse()

    items = await fetch_completed_annotations(queue_id)
    for item in items:
        # Check human score threshold
        scores = item.get("scores", [])
        quality_scores = [s["value"] for s in scores if s["name"] == "accuracy"]
        if quality_scores and quality_scores[0] >= 0.8:
            lf.create_dataset_item(
                dataset_name=dataset_name,
                trace_id=item["traceId"],
            )

Link to Golden Dataset Curation

Annotation queues feed directly into golden dataset curation:

1. Automated multi-agent pipeline scores content (accuracy, coherence, depth, relevance) 2. Items with quality_total >= 0.75 but low confidence go to the golden-dataset-candidates queue 3. Human reviewer confirms or overrides the automated decision 4. Approved traces are added to the evaluation dataset via create_dataset_item()

See ../../golden-dataset/rules/curation-annotation.md for the parallel multi-agent scoring pipeline that feeds this queue.

References

• Langfuse Annotation Queues docs
• ../references/evaluation-scores.md — automated scoring patterns
• ../../golden-dataset/rules/curation-annotation.md — multi-agent curation pipeline

Token & Cost Tracking

Automatic cost calculation based on model pricing, with spend alerts and Metrics API.

Basic Cost Tracking (v3)

from langfuse import observe, get_client, Langfuse

langfuse = Langfuse()

@observe(name="security_audit")
async def run_audit(content: str):
    """Track costs automatically via @observe."""
    response = await llm.generate(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": f"Analyze for XSS: {content}"}],
    )

    get_client().update_current_observation(
        model="claude-sonnet-4-6",
        usage={
            "input": 1500,
            "output": 1000,
            "unit": "TOKENS",
        },
    )
    # Langfuse automatically calculates: $0.0045 + $0.015 = $0.0195
    return response

Pricing Database (Auto-Updated)

Langfuse maintains a pricing database for all major models. You can also define custom pricing:

langfuse = Langfuse()

# Custom model pricing
langfuse.create_model(
    model_name="claude-sonnet-4-6",
    match_pattern="claude-sonnet-4.*",
    unit="TOKENS",
    input_price=0.000003,  # $3/MTok
    output_price=0.000015,  # $15/MTok
    total_price=None,  # Calculated from input+output
)

Cost Tracking Per Analysis

from langfuse import Langfuse

langfuse = Langfuse()

# After analysis completes
trace = langfuse.get_trace(trace_id)
total_cost = sum(
    gen.calculated_total_cost or 0
    for gen in trace.observations
    if gen.type == "GENERATION"
)

# Store in database
await analysis_repo.update(
    analysis_id,
    langfuse_trace_id=trace.id,
    total_cost_usd=total_cost,
)

Spend Alerts

Configure alerts to get notified when costs exceed thresholds:

In Langfuse UI

1. Navigate to Settings → Alerts 2. Create alert rule:

• Metric: Daily cost
• Threshold: $50/day
• Channel: Slack / Email / Webhook

Via API

langfuse = Langfuse()

# Programmatic spend check
from datetime import datetime, timedelta

# Get daily cost via v2 Metrics API
metrics = langfuse.get_metrics(
    metric_name="total_cost",
    from_timestamp=datetime.now() - timedelta(days=1),
    to_timestamp=datetime.now(),
)

daily_cost = metrics.values[0].value if metrics.values else 0

if daily_cost > 50.0:
    await send_alert(
        channel="slack",
        message=f"Daily LLM cost alert: ${daily_cost:.2f} exceeds $50 threshold",
    )

v2 Metrics API (Beta)

Query cost and usage metrics programmatically instead of SQL:

from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse()

# Total cost over last 7 days
metrics = langfuse.get_metrics(
    metric_name="total_cost",
    from_timestamp=datetime.now() - timedelta(days=7),
    to_timestamp=datetime.now(),
    granularity="day",
)

for point in metrics.values:
    print(f"{point.timestamp.date()}: ${point.value:.2f}")

# Token usage by model
token_metrics = langfuse.get_metrics(
    metric_name="total_tokens",
    from_timestamp=datetime.now() - timedelta(days=7),
    to_timestamp=datetime.now(),
    group_by="model",
)

for group in token_metrics.groups:
    print(f"{group.key}: {group.values[0].value:,} tokens")

Monitoring Dashboard Queries

Top 10 Most Expensive Traces (Last 7 Days)

SELECT
    name,
    user_id,
    calculated_total_cost,
    input_tokens,
    output_tokens
FROM traces
WHERE timestamp > NOW() - INTERVAL '7 days'
ORDER BY calculated_total_cost DESC
LIMIT 10;

Average Cost by Agent Type

SELECT
    metadata->>'agent_type' as agent,
    COUNT(*) as traces,
    AVG(calculated_total_cost) as avg_cost,
    SUM(calculated_total_cost) as total_cost
FROM traces
WHERE metadata->>'agent_type' IS NOT NULL
GROUP BY agent
ORDER BY total_cost DESC;

Daily Cost Trend

SELECT
    DATE(timestamp) as date,
    SUM(calculated_total_cost) as daily_cost,
    COUNT(*) as trace_count
FROM traces
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date;

Best Practices

1. Always pass usage data with input/output token counts 2. Monitor costs daily with spend alerts to catch spikes early 3. Set up threshold alerts for abnormal cost increases (> 2x daily average) 4. Track costs by user_id to identify expensive users 5. Group by metadata (content_type, agent_type) for cost attribution 6. Use custom pricing for self-hosted models 7. Use Metrics API for programmatic cost queries instead of raw SQL

References

• Langfuse Model Pricing
• Metrics API
• Spend Alerts

Monitoring Dashboards

Grafana dashboard patterns and SLO/SLI definitions.

The Four Golden Signals

SLO/SLI Examples

# SLO: 99.9% availability
SLI: availability = successful_requests / total_requests
Target: > 0.999

# SLO: 95% of requests < 500ms
SLI: latency_p95 = histogram_quantile(0.95, request_duration_seconds)
Target: < 0.5

# SLO: < 0.1% error rate
SLI: error_rate = failed_requests / total_requests
Target: < 0.001

Grafana Dashboard Structure

1. Overview row - traffic, errors, latency 2. Saturation row - CPU, memory, disk 3. Details row - per-endpoint breakdown 4. Database row - query performance, connections

Best Practices

1. Use time ranges - Last 1h, 6h, 24h, 7d 2. Percentiles over averages - p50, p95, p99 3. Color code thresholds - green/yellow/red 4. Include annotations - deployments, incidents

See Grafana dashboards in backend/grafana/dashboards/.

dev-agent-lens Integration

LiteLLM-based proxy that intercepts Claude API calls for cost tracking, latency monitoring, and model routing visibility. Complements OrchestKit's hook-level JSONL analytics with API-level observability.

When to Use dev-agent-lens vs Other Layers

Use dev-agent-lens when you need: per-request cost breakdown, model version tracking, API error rates, prompt/completion token ratios, latency percentiles at the API boundary.

Don't use dev-agent-lens when: you only need hook/skill/agent-level data (use JSONL), or tool-level spans (use CC OTLP).

Architecture

Claude Code CLI
  │
  ├─ ANTHROPIC_BASE_URL=http://localhost:4000 ──→ dev-agent-lens (LiteLLM proxy :4000)
  │                                                  │
  │                                                  ├─→ Anthropic API (actual model calls)
  │                                                  ├─→ Langfuse (traces, costs)
  │                                                  └─→ Prometheus (:9090, optional)
  │
  ├─ OTEL_EXPORTER_OTLP_ENDPOINT ──→ Langfuse OTEL (:3100/api/public/otel)
  │                                    (tool-level spans from CC native telemetry)
  │
  └─ ~/.claude/analytics/*.jsonl ──→ JSONL bridge script (optional)
                                      (hook timing, agent routing, skill usage)

API Key Caveat

Claude Code Free/Pro users: Cannot use ANTHROPIC_BASE_URL — the CLI sends requests directly to Anthropic's API using your subscription. The proxy approach only works with API key access (pay-per-token via ANTHROPIC_API_KEY).

Claude Code Max users: Same limitation — Max plans route through Anthropic's managed infrastructure, not a configurable base URL.

This means dev-agent-lens is primarily useful for:

• Self-hosted/enterprise deployments using API keys
• Development environments where you control the API routing
• CI/CD pipelines calling Claude via API

Docker Compose Template

# Add to your project's docker-compose.yml
# Profile: observability (docker compose --profile observability up)

services:
  dev-agent-lens:
    image: ghcr.io/berriai/litellm:main-latest
    profiles: [observability]
    ports:
      - "4000:4000"
    environment:
      LITELLM_MASTER_KEY: "sk-dev-local"
      LANGFUSE_PUBLIC_KEY: "${LANGFUSE_PUBLIC_KEY:-pk-lf-dev}"
      LANGFUSE_SECRET_KEY: "${LANGFUSE_SECRET_KEY:-sk-lf-dev}"
      LANGFUSE_HOST: "${LANGFUSE_HOST:-http://langfuse-web:3100}"
    volumes:
      - ./litellm-config.yaml:/app/config.yaml
    command: ["--config", "/app/config.yaml"]
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

LiteLLM Config (litellm-config.yaml)

model_list:
  - model_name: claude-sonnet-4-6
    litellm_params:
      model: anthropic/claude-sonnet-4-6
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: claude-opus-4-8
    litellm_params:
      model: anthropic/claude-opus-4-8
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: claude-haiku-4-5-20251001
    litellm_params:
      model: anthropic/claude-haiku-4-5-20251001
      api_key: os.environ/ANTHROPIC_API_KEY

general_settings:
  master_key: sk-dev-local

litellm_settings:
  success_callback: ["langfuse"]
  failure_callback: ["langfuse"]
  cache: false
  set_verbose: false

Shell Configuration (API key users only)

# ~/.zshrc or ~/.bashrc — only for API key access
export ANTHROPIC_BASE_URL="http://localhost:4000"
export ANTHROPIC_API_KEY="sk-ant-..."  # your actual key

What You See in Langfuse

With dev-agent-lens forwarding to Langfuse, each Claude API call creates a trace with:

• Model: Exact model ID (claude-sonnet-4-6)
• Tokens: Input/output/cache token counts
• Cost: Per-request USD cost (Anthropic pricing)
• Latency: Time-to-first-token, total duration
• Status: Success/failure, error codes, rate limits
• Metadata: Request headers, retry counts

This is the API boundary layer — it sees what crosses the network. It does NOT see:

• Which OrchestKit agent spawned the request (use JSONL for that)
• Which tool CC executed (use CC OTLP for that)
• Hook execution timing (use JSONL for that)

Complementary Data: Proxy + Hooks + OTLP

The three layers together give full observability:

Question: "Why was this session slow?"

Layer 1 (CC OTLP):    Tool spans show 47 Read calls, 12 Bash calls
Layer 2 (JSONL):       Hook timing shows pre-push hook took 8.3s
Layer 3 (Proxy):       API calls show 3 rate-limited retries, p99 latency 4.2s

Answer: Rate limiting + excessive file reads + slow pre-push hook

References

• LiteLLM Proxy docs
• Langfuse LiteLLM integration
• CC OTLP telemetry

Distributed Tracing

Track requests across microservices with OpenTelemetry.

Basic Setup (Node.js)

import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';

const sdk = new NodeSDK({
  traceExporter: new JaegerExporter(),
  instrumentations: [getNodeAutoInstrumentations()],
});

sdk.start();

Span Relationships

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

# Parent span
with tracer.start_as_current_span("analyze_content") as parent_span:
    parent_span.set_attribute("content.url", url)
    parent_span.set_attribute("content.type", "article")

    # Child span (sequential)
    with tracer.start_as_current_span("fetch_content") as fetch_span:
        content = await fetch_url(url)
        fetch_span.set_attribute("content.size_bytes", len(content))

    # Another child span (sequential)
    with tracer.start_as_current_span("generate_embedding") as embed_span:
        embedding = await embed_text(content)
        embed_span.set_attribute("embedding.dimensions", len(embedding))

    # Parallel child spans (using asyncio.gather)
    async def analyze_with_span(agent_name: str, content: str):
        with tracer.start_as_current_span(f"agent_{agent_name}"):
            return await agent.analyze(content)

    results = await asyncio.gather(
        analyze_with_span("tech_comparator", content),
        analyze_with_span("security_auditor", content),
        analyze_with_span("implementation_planner", content)
    )

Trace Sampling Strategies

Head-based sampling (decide at trace start):

from opentelemetry.sdk.trace.sampling import (
    TraceIdRatioBased,  # Sample X% of traces
    ParentBased,        # Follow parent's sampling decision
    ALWAYS_ON,          # Always sample
    ALWAYS_OFF          # Never sample
)

# Sample 10% of traces
sampler = TraceIdRatioBased(0.1)

Tail-based sampling (decide after trace completes):

• Keep all traces with errors
• Keep slow traces (p95+ latency)
• Sample 1% of successful fast traces

Recommended sampling:

• Development: 100% sampling
• Production: 10% sampling, 100% for errors

Context Propagation

// Service A: Create trace context
const ctx = context.active();

// Service B: Extract trace context from headers
const propagatedCtx = propagation.extract(ctx, request.headers);
context.with(propagatedCtx, () => {
  // This span will be child of Service A's span
  const span = tracer.startSpan('service_b_operation');
  // ...
  span.end();
});

Trace Analysis Queries

Find slow traces:

duration > 2s

Find traces with errors:

status = error

Find traces for specific user:

user.id = "abc-123"

Find traces hitting specific service:

service.name = "analysis-worker"

Claude Code TRACEPARENT Propagation (CC 2.1.97)

CC 2.1.97 injects a W3C TRACEPARENT env var into all Bash subprocesses when OTEL tracing is enabled. This enables end-to-end distributed tracing from Claude Code through to your services.

Format: TRACEPARENT=00-{trace_id}-{parent_id}-{trace_flags}

// In a subprocess spawned by Claude Code's Bash tool:
const traceparent = process.env.TRACEPARENT;
if (traceparent) {
  // Parse W3C Trace Context header
  const [version, traceId, parentId, traceFlags] = traceparent.split('-');

  // Propagate to downstream HTTP calls
  fetch('https://api.example.com/data', {
    headers: { 'traceparent': traceparent },
  });
}

OrchestKit hook telemetry correlation: Hook events forwarded to HQ include the TRACEPARENT value when available, enabling correlation between CC tool spans and downstream service traces in Langfuse/Jaeger.

// Telemetry payload includes traceparent for cross-system correlation
{
  "event": "PostToolUse",
  "tool": "Bash",
  "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
  "session_id": "...",
  "timestamp": "..."
}

Best Practices

1. Sample smartly - 10% for high traffic, 100% for errors 2. Add attributes - user_id, order_id, error_type 3. Propagate context - across HTTP, gRPC, message queues 4. Tag errors - error=true for filtering 5. Capture TRACEPARENT - in subprocesses spawned by CC for end-to-end traces (CC 2.1.97)

See scripts/opentelemetry-tracing.ts for complete setup.

Embedding Drift Detection

Monitor semantic drift in LLM applications using embedding-based methods.

Overview

Traditional statistical methods (PSI, KS) don't work well for unstructured text data. Embedding drift detection uses vector representations to detect semantic changes.

Arize Phoenix Integration

import phoenix as px
from phoenix.trace import TraceDataset
import numpy as np

# Launch Phoenix for local observability
session = px.launch_app()

# Analyze embedding drift
def analyze_embedding_drift(
    baseline_embeddings: np.ndarray,
    current_embeddings: np.ndarray
) -> dict:
    """
    Analyze drift in embedding space using Phoenix.

    Args:
        baseline_embeddings: Reference embeddings (N x D)
        current_embeddings: Current embeddings (M x D)

    Returns:
        Drift analysis results
    """
    # Phoenix provides built-in drift analysis
    drift_analysis = px.Client().compute_drift(
        primary_embeddings=current_embeddings,
        reference_embeddings=baseline_embeddings
    )

    return drift_analysis

Cluster-Based Drift Detection

from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
import numpy as np

class ClusterDriftDetector:
    """Detect drift by monitoring cluster distributions."""

    def __init__(self, n_clusters: int = 10, psi_threshold: float = 0.25):
        self.n_clusters = n_clusters
        self.psi_threshold = psi_threshold
        self.kmeans = None
        self.baseline_distribution = None

    def fit_baseline(self, embeddings: np.ndarray):
        """Fit clusters on baseline embeddings."""
        self.kmeans = KMeans(
            n_clusters=self.n_clusters,
            random_state=42,
            n_init=10
        )
        labels = self.kmeans.fit_predict(embeddings)

        # Store baseline cluster distribution
        self.baseline_distribution = np.bincount(
            labels,
            minlength=self.n_clusters
        ) / len(labels)

        return self

    def detect_drift(self, embeddings: np.ndarray) -> dict:
        """Detect drift in new embeddings."""
        if self.kmeans is None:
            raise ValueError("Must call fit_baseline first")

        # Assign new embeddings to clusters
        labels = self.kmeans.predict(embeddings)

        # Current cluster distribution
        current_distribution = np.bincount(
            labels,
            minlength=self.n_clusters
        ) / len(labels)

        # Calculate PSI between distributions
        psi = self._calculate_psi(
            self.baseline_distribution,
            current_distribution
        )

        # Calculate centroid distances
        centroid_shift = self._calculate_centroid_shift(embeddings, labels)

        return {
            "psi": psi,
            "drift_detected": psi > self.psi_threshold,
            "baseline_distribution": self.baseline_distribution.tolist(),
            "current_distribution": current_distribution.tolist(),
            "centroid_shift": centroid_shift,
            "interpretation": self._interpret(psi, centroid_shift)
        }

    def _calculate_psi(self, expected: np.ndarray, actual: np.ndarray) -> float:
        """Calculate PSI between cluster distributions."""
        eps = 0.0001
        expected = expected + eps
        actual = actual + eps
        return np.sum((actual - expected) * np.log(actual / expected))

    def _calculate_centroid_shift(
        self,
        embeddings: np.ndarray,
        labels: np.ndarray
    ) -> dict:
        """Calculate how much cluster centroids have shifted."""
        shifts = {}
        for i in range(self.n_clusters):
            cluster_embeddings = embeddings[labels == i]
            if len(cluster_embeddings) > 0:
                current_centroid = cluster_embeddings.mean(axis=0)
                baseline_centroid = self.kmeans.cluster_centers_[i]
                shift = np.linalg.norm(current_centroid - baseline_centroid)
                shifts[f"cluster_{i}"] = float(shift)
        return shifts

    def _interpret(self, psi: float, centroid_shift: dict) -> str:
        avg_shift = np.mean(list(centroid_shift.values()))
        if psi < 0.1 and avg_shift < 0.1:
            return "No significant drift"
        elif psi < 0.25:
            return "Minor drift detected, monitor closely"
        else:
            return "Significant drift, investigate and consider retraining"

Centroid Distance Monitoring

import numpy as np
from typing import Optional

class CentroidMonitor:
    """Monitor drift via embedding centroid movement."""

    def __init__(self, distance_threshold: float = 0.2):
        self.distance_threshold = distance_threshold
        self.baseline_centroid: Optional[np.ndarray] = None
        self.baseline_std: Optional[float] = None

    def set_baseline(self, embeddings: np.ndarray):
        """Set baseline centroid from reference embeddings."""
        self.baseline_centroid = embeddings.mean(axis=0)

        # Calculate average distance from centroid
        distances = np.linalg.norm(
            embeddings - self.baseline_centroid,
            axis=1
        )
        self.baseline_std = distances.std()

        return self

    def check_drift(self, embeddings: np.ndarray) -> dict:
        """Check if current embeddings have drifted from baseline."""
        if self.baseline_centroid is None:
            raise ValueError("Must call set_baseline first")

        # Current centroid
        current_centroid = embeddings.mean(axis=0)

        # Distance between centroids
        centroid_distance = np.linalg.norm(
            current_centroid - self.baseline_centroid
        )

        # Normalized by baseline spread
        normalized_distance = centroid_distance / (self.baseline_std + 1e-10)

        # Check individual embedding distances
        distances = np.linalg.norm(
            embeddings - self.baseline_centroid,
            axis=1
        )
        outlier_ratio = (distances > 3 * self.baseline_std).mean()

        return {
            "centroid_distance": float(centroid_distance),
            "normalized_distance": float(normalized_distance),
            "outlier_ratio": float(outlier_ratio),
            "drift_detected": normalized_distance > self.distance_threshold,
            "severity": self._severity(normalized_distance, outlier_ratio)
        }

    def _severity(self, distance: float, outlier_ratio: float) -> str:
        if distance < 0.1 and outlier_ratio < 0.05:
            return "none"
        elif distance < 0.2 and outlier_ratio < 0.1:
            return "low"
        elif distance < 0.3 and outlier_ratio < 0.2:
            return "medium"
        else:
            return "high"

Cosine Similarity Drift

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def cosine_drift_score(
    baseline_embeddings: np.ndarray,
    current_embeddings: np.ndarray,
    sample_size: int = 1000
) -> dict:
    """
    Measure drift using cosine similarity distributions.

    Args:
        baseline_embeddings: Reference embeddings
        current_embeddings: Current embeddings
        sample_size: Number of pairs to sample

    Returns:
        Drift analysis based on cosine similarities
    """
    # Sample pairs for efficiency
    n_baseline = min(len(baseline_embeddings), sample_size)
    n_current = min(len(current_embeddings), sample_size)

    baseline_sample = baseline_embeddings[
        np.random.choice(len(baseline_embeddings), n_baseline, replace=False)
    ]
    current_sample = current_embeddings[
        np.random.choice(len(current_embeddings), n_current, replace=False)
    ]

    # Baseline self-similarity
    baseline_centroid = baseline_sample.mean(axis=0)
    baseline_similarities = cosine_similarity(
        baseline_sample,
        baseline_centroid.reshape(1, -1)
    ).flatten()

    # Current similarity to baseline centroid
    current_similarities = cosine_similarity(
        current_sample,
        baseline_centroid.reshape(1, -1)
    ).flatten()

    # Compare distributions
    baseline_mean = baseline_similarities.mean()
    current_mean = current_similarities.mean()

    similarity_drop = baseline_mean - current_mean

    return {
        "baseline_mean_similarity": float(baseline_mean),
        "current_mean_similarity": float(current_mean),
        "similarity_drop": float(similarity_drop),
        "drift_detected": similarity_drop > 0.1,
        "interpretation": (
            "Significant semantic drift"
            if similarity_drop > 0.1
            else "No significant drift"
        )
    }

RAG Retrieval Drift

from typing import List
import numpy as np

class RAGDriftMonitor:
    """Monitor drift in RAG retrieval quality."""

    def __init__(
        self,
        similarity_threshold: float = 0.7,
        coverage_threshold: float = 0.8
    ):
        self.similarity_threshold = similarity_threshold
        self.coverage_threshold = coverage_threshold
        self.baseline_queries: List[np.ndarray] = []
        self.baseline_retrievals: List[List[np.ndarray]] = []

    def add_baseline(
        self,
        query_embedding: np.ndarray,
        retrieved_embeddings: List[np.ndarray]
    ):
        """Add a query-retrieval pair to baseline."""
        self.baseline_queries.append(query_embedding)
        self.baseline_retrievals.append(retrieved_embeddings)

    def check_retrieval_drift(
        self,
        query_embedding: np.ndarray,
        retrieved_embeddings: List[np.ndarray]
    ) -> dict:
        """
        Check if retrieval for a query has drifted.

        Useful for detecting:
        - Index staleness
        - Embedding model changes
        - Document corpus drift
        """
        # Find most similar baseline query
        similarities = [
            cosine_similarity(
                query_embedding.reshape(1, -1),
                bq.reshape(1, -1)
            )[0, 0]
            for bq in self.baseline_queries
        ]

        best_match_idx = np.argmax(similarities)
        query_similarity = similarities[best_match_idx]

        if query_similarity < self.similarity_threshold:
            return {
                "drift_detected": False,
                "reason": "Query too different from baseline"
            }

        # Compare retrieved documents
        baseline_retrieved = self.baseline_retrievals[best_match_idx]

        # Calculate coverage: how many baseline docs are still retrieved
        coverage = self._calculate_coverage(
            baseline_retrieved,
            retrieved_embeddings
        )

        return {
            "query_similarity": float(query_similarity),
            "coverage": float(coverage),
            "drift_detected": coverage < self.coverage_threshold,
            "interpretation": (
                f"Retrieval coverage dropped to {coverage:.2%}"
                if coverage < self.coverage_threshold
                else "Retrieval stable"
            )
        }

    def _calculate_coverage(
        self,
        baseline: List[np.ndarray],
        current: List[np.ndarray]
    ) -> float:
        """Calculate what fraction of baseline docs are still retrieved."""
        if not baseline or not current:
            return 0.0

        baseline_stack = np.stack(baseline)
        current_stack = np.stack(current)

        # For each baseline doc, check if similar doc is in current
        similarities = cosine_similarity(baseline_stack, current_stack)
        max_similarities = similarities.max(axis=1)

        # Count docs with similarity > threshold
        covered = (max_similarities > self.similarity_threshold).sum()

        return covered / len(baseline)

Evidently AI Integration

from evidently import Report
from evidently.metrics import EmbeddingsDriftMetric
import pandas as pd
import numpy as np

def evidently_embedding_drift(
    baseline_embeddings: np.ndarray,
    current_embeddings: np.ndarray,
    embedding_column: str = "embedding"
) -> dict:
    """
    Use Evidently AI for embedding drift detection.

    Evidently uses model-based drift detection by default:
    Trains a classifier to distinguish baseline vs current.
    """
    # Create DataFrames
    baseline_df = pd.DataFrame({
        embedding_column: list(baseline_embeddings)
    })
    current_df = pd.DataFrame({
        embedding_column: list(current_embeddings)
    })

    # Run Evidently report
    report = Report(metrics=[
        EmbeddingsDriftMetric(column_name=embedding_column)
    ])

    report.run(
        reference_data=baseline_df,
        current_data=current_df
    )

    # Extract results
    result = report.as_dict()["metrics"][0]["result"]

    return {
        "drift_score": result.get("drift_score"),
        "drift_detected": result.get("drift_detected"),
        "method": "model_based",
        "details": result
    }

References

• Arize Phoenix Documentation
• Evidently AI: Embedding Drift Detection
• Measuring Embedding Drift
• AWS: Monitor Embedding Drift for LLMs

LLM Evaluation & Scoring

Track quality metrics with custom scores, automated evaluation, and evaluator execution tracing.

Basic Scoring (v3)

from langfuse import observe, get_client, Langfuse

langfuse = Langfuse()

@observe()
async def analyze_and_score(query: str):
    """Run analysis and score the result."""
    response = await llm.generate(query)

    # Score via get_client() within @observe context
    get_client().update_current_observation(
        output=response[:500],
    )

    # Score the trace
    get_client().score_current_trace(
        name="relevance",
        value=0.85,
        comment="Response addresses query but lacks depth",
    )
    return response


# Or score by trace_id directly
langfuse.create_score(
    trace_id="trace_123",
    name="factuality",
    value=0.92,
    data_type="NUMERIC",
)

Evaluator Execution Tracing

In v3, each evaluator run creates its own inspectable trace:

from langfuse import observe, get_client

@observe(as_type="evaluator", name="relevance_judge")
async def evaluate_relevance(query: str, response: str):
    """Each evaluator call creates an inspectable trace in Langfuse."""
    score = await llm_judge.evaluate(
        criteria="relevance",
        query=query,
        response=response,
    )

    get_client().update_current_observation(
        input={"query": query[:500], "response": response[:500]},
        output={"score": score, "criteria": "relevance"},
        model="claude-sonnet-4-6",
    )

    # The evaluator's own LLM calls are visible in its trace
    return score

Result in Langfuse UI:

evaluator:relevance_judge (0.8s, $0.01)
├── generation: judge_prompt → score: 0.85
└── metadata: {criteria: "relevance", model: "claude-sonnet-4-6"}

Score Analytics

View multi-score comparisons in the Langfuse dashboard:

• Score distributions: Histogram of scores by criterion
• Multi-score comparison: Side-by-side comparison of relevance, depth, accuracy
• Quality trends: Track scores over time
• Filter by threshold: Show only low-scoring traces
• Compare prompts: Which prompt version scores higher?

Mutable Score Configs

Configure score types and ranges in Langfuse settings:

# Score configs can be updated without code changes
# In Langfuse UI: Settings → Score Configs

# Numeric scores
langfuse.create_score(trace_id="...", name="relevance", value=0.85, data_type="NUMERIC")

# Categorical scores
langfuse.create_score(trace_id="...", name="sentiment", value="positive", data_type="CATEGORICAL")

# Boolean scores
langfuse.create_score(trace_id="...", name="contains_pii", value=0, data_type="BOOLEAN")

Automated Scoring with G-Eval

from langfuse import observe, get_client
from app.shared.services.g_eval import GEvalScorer

scorer = GEvalScorer()

@observe()
async def analyze_with_scoring(query: str):
    response = await llm.generate(query)

    # Run G-Eval scoring
    scores = await scorer.score(
        query=query,
        response=response,
        criteria=["relevance", "coherence", "depth"],
    )

    # Record all scores
    for criterion, score in scores.items():
        get_client().score_current_trace(name=criterion, value=score)

    return response

Quality Scores Trend Query

SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'relevance') as avg_relevance,
    AVG(value) FILTER (WHERE name = 'depth') as avg_depth,
    AVG(value) FILTER (WHERE name = 'factuality') as avg_factuality
FROM scores
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date;

Datasets for Evaluation

Create test datasets and run automated evaluations:

from langfuse import Langfuse, observe, get_client

langfuse = Langfuse()

# Fetch dataset
dataset = langfuse.get_dataset("security_audit_test_set")

@observe()
async def evaluate_item(item):
    """Evaluate a single dataset item with tracing."""
    response = await llm.generate(item.input)

    get_client().update_current_observation(
        input=item.input,
        output=response,
    )

    # Score
    score = await evaluate_response(item.expected_output, response)
    get_client().score_current_trace(name="accuracy", value=score)

    return score

# Run evaluation
for item in dataset.items:
    await evaluate_item(item)

Dataset Structure in UI

security_audit_test_set
├── item_1: XSS vulnerability test
│   ├── input: "Check this HTML for XSS..."
│   └── expected_output: "Found XSS in innerHTML..."
├── item_2: SQL injection test
│   ├── input: "Review this SQL query..."
│   └── expected_output: "SQL injection vulnerability in WHERE clause..."
└── item_3: CSRF protection test
    ├── input: "Analyze this form..."
    └── expected_output: "Missing CSRF token..."

Evaluation Metrics

Common score types:

Best Practices

1. Score all production traces for quality monitoring 2. Use evaluator type (@observe(as_type="evaluator")) for inspectable judge traces 3. Use consistent criteria across all evaluations 4. Automate scoring with G-Eval or similar 5. Set quality thresholds (e.g., avg_relevance > 0.7) 6. Create test datasets for regression testing 7. Track scores by prompt version to measure improvements 8. Alert on quality drops (e.g., avg_score < 0.6 for 3 days)

Integration with OrchestKit Quality Gate

from langfuse import observe, get_client

@observe(name="quality_gate")
async def quality_gate_node(state: WorkflowState):
    """Quality gate with Langfuse scoring."""

    # Get scores from evaluators
    scores = await run_quality_evaluators(state)

    # Log scores to trace
    for criterion, score in scores.items():
        get_client().score_current_trace(name=criterion, value=score)

    # Check threshold
    avg_score = sum(scores.values()) / len(scores)
    passed = avg_score >= 0.7

    return {"quality_gate_passed": passed, "quality_scores": scores}

References

• Langfuse Scores
• Evaluator Tracing
• Score Analytics
• Datasets Guide

EWMA Dynamic Baselines

Exponentially Weighted Moving Average for adaptive drift detection baselines.

Basic EWMA

import numpy as np
from dataclasses import dataclass

@dataclass
class EWMAState:
    mean: float = 0.0
    variance: float = 0.0
    count: int = 0

class EWMABaseline:
    """EWMA-based dynamic baseline. Formula: EWMA_t = α × X_t + (1-α) × EWMA_{t-1}"""

    def __init__(self, alpha: float = 0.2, sigma_threshold: float = 3.0, min_samples: int = 10):
        self.alpha = alpha
        self.sigma_threshold = sigma_threshold
        self.min_samples = min_samples
        self.state = EWMAState()

    def update(self, value: float) -> dict:
        """Update baseline and check for anomaly."""
        self.state.count += 1

        if self.state.count == 1:
            self.state.mean = value
            self.state.variance = 0.0
        else:
            delta = value - self.state.mean
            self.state.mean = self.alpha * value + (1 - self.alpha) * self.state.mean
            self.state.variance = (1 - self.alpha) * (self.state.variance + self.alpha * delta ** 2)

        std = np.sqrt(self.state.variance) if self.state.variance > 0 else 0.001
        z_score = abs(value - self.state.mean) / std
        is_anomaly = self.state.count >= self.min_samples and z_score > self.sigma_threshold

        return {
            "value": value,
            "ewma_mean": self.state.mean,
            "ewma_std": std,
            "z_score": z_score,
            "is_anomaly": is_anomaly
        }

Multi-Metric Tracker

class MultiMetricEWMA:
    """Track multiple metrics with independent baselines."""

    def __init__(self, metrics: list[str], alpha: float = 0.2):
        self.baselines = {m: EWMABaseline(alpha=alpha) for m in metrics}

    def update(self, metrics: dict) -> dict:
        results = {}
        anomalies = []
        for name, value in metrics.items():
            if name in self.baselines:
                result = self.baselines[name].update(value)
                results[name] = result
                if result["is_anomaly"]:
                    anomalies.append({"metric": name, "z_score": result["z_score"]})
        return {"metrics": results, "anomalies": anomalies}

Alpha Selection

References

• EWMA Control Charts

Langfuse + Evidently AI Integration

Combining Langfuse tracing with Evidently AI drift detection.

Export Langfuse Data

from langfuse import Langfuse
import pandas as pd
from datetime import datetime, timedelta

langfuse = Langfuse()

def export_langfuse_scores(days: int = 7) -> pd.DataFrame:
    """Export Langfuse scores to DataFrame for Evidently."""
    traces = langfuse.get_traces(from_timestamp=datetime.now() - timedelta(days=days))
    records = []
    for trace in traces.data:
        scores = {s.name: s.value for s in trace.scores}
        if scores:
            records.append({"trace_id": trace.id, "timestamp": trace.timestamp, **scores})
    return pd.DataFrame(records)

Evidently Drift Report

from evidently import Report
from evidently.metrics import DatasetDriftMetric, ColumnDriftMetric

def run_drift_report(baseline_df: pd.DataFrame, current_df: pd.DataFrame, columns: list) -> dict:
    """Run Evidently drift detection."""
    report = Report(metrics=[DatasetDriftMetric()])
    for col in columns:
        report.metrics.append(ColumnDriftMetric(column_name=col))

    report.run(reference_data=baseline_df, current_data=current_df)
    result = report.as_dict()

    return {
        "dataset_drift": result["metrics"][0]["result"].get("dataset_drift"),
        "drift_share": result["metrics"][0]["result"].get("share_of_drifted_columns")
    }

Automated Monitoring

class LangfuseEvidentlyMonitor:
    def __init__(self, baseline_days: int = 7, current_days: int = 1):
        self.langfuse = Langfuse()
        self.baseline_days = baseline_days
        self.current_days = current_days

    def run_analysis(self, metrics: list) -> dict:
        baseline_df = export_langfuse_scores(self.baseline_days + self.current_days)
        current_df = export_langfuse_scores(self.current_days)
        results = run_drift_report(baseline_df, current_df, metrics)
        return results

References

• Evidently AI Docs
• Langfuse SDK

Structured Logging

JSON logging best practices for production systems.

Why Structured Logging?

• Searchable - query by fields (user_id, trace_id)
• Machine-readable - parse and aggregate easily
• Contextual - attach metadata to every log

Python (structlog)

import structlog

logger = structlog.get_logger()

logger.info("user_login", user_id="123", ip="192.168.1.1")
# Output: {"event": "user_login", "user_id": "123", "ip": "192.168.1.1", "timestamp": "2025-12-19T10:00:00Z"}

Node.js (pino)

import pino from 'pino';

const logger = pino();

logger.info({ userId: '123', action: 'login' }, 'User logged in');
// Output: {"level":30,"userId":"123","action":"login","msg":"User logged in","time":1702990800000}

Log Levels

Best Practices

1. Always include trace_id - correlate across services 2. Log at boundaries - API requests/responses, DB queries 3. Don't log secrets - mask passwords, API keys 4. Use correlation IDs - track requests across microservices

See scripts/structured-logging.ts for implementation.

[Rule Name]

[Brief description — 1-2 sentences.]

Incorrect:

// Bad pattern

Correct:

// Good pattern

Key rules:

• [Rule 1]
• [Rule 2]
• [Rule 3]

Reference: [link]