Analyze Observability Metrics with the Agenta Analytics API

Open in Google Colaboratory

Overview

The Agenta Analytics API retrieves aggregated metrics from your LLM traces. The API groups your data into time buckets and calculates metrics like cost, latency, token usage, and error rates.

Use analytics to:

• Track LLM costs over time
• Monitor performance trends
• Identify error patterns
• Analyze token consumption

Endpoint: POST /api/preview/tracing/spans/analytics

Authentication: You can create API keys from the Settings page in your Agenta workspace.

Quick Start

Python

import requests
from datetime import datetime, timedelta, timezone

# Setup
AGENTA_HOST = "https://cloud.agenta.ai"
API_KEY = "your_api_key_here"
BASE_URL = f"{AGENTA_HOST}/api/preview/tracing/spans/analytics"

headers = {
    "Authorization": f"ApiKey {API_KEY}",
    "Content-Type": "application/json"
}

# Get analytics for last 7 days with daily buckets
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=7)

payload = {
    "focus": "trace",
    "interval": 1440,  # 1440 minutes = daily buckets
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

print(f"Found {data['count']} daily buckets")
for bucket in data['buckets']:
    if bucket['total']['count'] > 0:
        print(f"\nDate: {bucket['timestamp'][:10]}")
        print(f"  Traces: {bucket['total']['count']}")
        print(f"  Cost: ${bucket['total']['costs']:.4f}")
        print(f"  Avg Duration: {bucket['total']['duration'] / bucket['total']['count']:.0f}ms")
        print(f"  Errors: {bucket['errors']['count']}")

JavaScript

const AGENTA_HOST = "https://cloud.agenta.ai";
const API_KEY = "your_api_key_here";
const BASE_URL = `${AGENTA_HOST}/api/preview/tracing/spans/analytics`;

// Get analytics for last 7 days with daily buckets
const newest = new Date();
const oldest = new Date(newest.getTime() - 7 * 24 * 60 * 60 * 1000);

const payload = {
  focus: "trace",
  interval: 1440, // 1440 minutes = daily buckets
  windowing: {
    oldest: oldest.toISOString(),
    newest: newest.toISOString()
  }
};

const response = await fetch(BASE_URL, {
  method: "POST",
  headers: {
    "Authorization": `ApiKey ${API_KEY}`,
    "Content-Type": "application/json"
  },
  body: JSON.stringify(payload)
});

const data = await response.json();

console.log(`Found ${data.count} daily buckets`);
data.buckets.forEach(bucket => {
  if (bucket.total.count > 0) {
    const date = bucket.timestamp.substring(0, 10);
    const avgDuration = bucket.total.duration / bucket.total.count;
    console.log(`\nDate: ${date}`);
    console.log(`  Traces: ${bucket.total.count}`);
    console.log(`  Cost: $${bucket.total.costs.toFixed(4)}`);
    console.log(`  Avg Duration: ${avgDuration.toFixed(0)}ms`);
    console.log(`  Errors: ${bucket.errors.count}`);
  }
});

Request Parameters

All parameters are sent in the JSON request body:

Focus (Required)

Controls the aggregation level:

• trace: Aggregate by complete traces (most common)
• span: Aggregate individual spans

Most analytics queries use trace to analyze complete LLM requests.

payload = {
    "focus": "trace",
    "interval": 1440
}

Interval (Required)

Bucket size in minutes. Common values:

• 60 = Hourly buckets
• 1440 = Daily buckets (24 hours)
• 10080 = Weekly buckets (7 days)

payload = {
    "focus": "trace",
    "interval": 1440  # Daily buckets
}

Windowing (Optional)

Specify the time range for your analytics. If not provided, defaults to the last 30 days.

from datetime import datetime, timedelta, timezone

newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=30)

payload = {
    "focus": "trace",
    "interval": 1440,
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

Filter (Optional)

Filter which traces to include in analytics. Uses the same filter syntax as the Query API.

Filter by status code:

payload = {
    "focus": "trace",
    "interval": 1440,
    "filter": {
        "conditions": [
            {
                "field": "status.code",
                "operator": "eq",
                "value": "STATUS_CODE_OK"
            }
        ]
    }
}

Filter by span name:

payload = {
    "focus": "trace",
    "interval": 1440,
    "filter": {
        "conditions": [
            {
                "field": "name",
                "operator": "contains",
                "value": "openai"
            }
        ]
    }
}

Multiple conditions:

payload = {
    "focus": "trace",
    "interval": 1440,
    "filter": {
        "operator": "and",
        "conditions": [
            {
                "field": "status.code",
                "operator": "eq",
                "value": "STATUS_CODE_OK"
            },
            {
                "field": "attributes.ag.metrics.costs.cumulative.total",
                "operator": "gt",
                "value": 0.01
            }
        ]
    }
}

Response Format

The API returns aggregated metrics grouped into time buckets:

{
  "count": 7,
  "buckets": [
    {
      "timestamp": "2025-10-24T00:00:00Z",
      "interval": 1440,
      "total": {
        "count": 150,
        "duration": 45000.5,
        "costs": 0.0234,
        "tokens": 1200.0
      },
      "errors": {
        "count": 5,
        "duration": 2300.0,
        "costs": 0.0,
        "tokens": 0.0
      }
    },
    {
      "timestamp": "2025-10-25T00:00:00Z",
      "interval": 1440,
      "total": {
        "count": 200,
        "duration": 60000.0,
        "costs": 0.0312,
        "tokens": 1600.0
      },
      "errors": {
        "count": 3,
        "duration": 1500.0,
        "costs": 0.0,
        "tokens": 0.0
      }
    }
  ]
}

Response Fields

• count: Number of time buckets returned
• buckets: Array of time-based aggregated metrics

Bucket Fields

• timestamp: Start time of the bucket (ISO 8601)
• interval: Bucket size in minutes
• total: Aggregated metrics for all traces in this bucket
  • count: Number of traces
  • duration: Total duration in milliseconds
  • costs: Total cost in USD
  • tokens: Total tokens used
• errors: Aggregated metrics for failed traces only
  • count: Number of failed traces
  • duration: Total duration of failed traces in milliseconds
  • costs: Total cost of failed traces (usually 0)
  • tokens: Total tokens in failed traces (usually 0)

Common Use Cases

Monitor Daily Costs

Track LLM spending over time:

import requests
from datetime import datetime, timedelta, timezone

# Get daily costs for last 30 days
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=30)

payload = {
    "focus": "trace",
    "interval": 1440,  # Daily buckets
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

# Calculate totals
total_cost = sum(b['total']['costs'] for b in data['buckets'])
total_traces = sum(b['total']['count'] for b in data['buckets'])

print(f"Total Cost (30 days): ${total_cost:.2f}")
print(f"Total Traces: {total_traces:,}")
print(f"Average Cost per Trace: ${total_cost/total_traces:.4f}")

Analyze Error Trends

Monitor error rates to identify reliability issues:

# Get hourly metrics for last 7 days
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=7)

payload = {
    "focus": "trace",
    "interval": 60,  # Hourly buckets
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

# Find high error rate periods
print("Hours with high error rates (>5%):")
for bucket in data['buckets']:
    if bucket['total']['count'] > 0:
        error_rate = (bucket['errors']['count'] / bucket['total']['count']) * 100
        if error_rate > 5:
            print(f"  {bucket['timestamp']}: {error_rate:.1f}%")

Track Token Usage

Monitor token consumption patterns:

# Get daily token usage for last 7 days
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=7)

payload = {
    "focus": "trace",
    "interval": 1440,  # Daily buckets
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

print("Daily Token Usage:")
for bucket in data['buckets']:
    if bucket['total']['count'] > 0:
        date = bucket['timestamp'][:10]
        avg_tokens = bucket['total']['tokens'] / bucket['total']['count']
        print(f"  {date}: {bucket['total']['tokens']:,.0f} total ({avg_tokens:.0f} avg)")

Compare Performance

Analyze latency trends over time:

# Get hourly performance for last 24 hours
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=1)

payload = {
    "focus": "trace",
    "interval": 60,  # Hourly buckets
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

print("Hourly Average Latency:")
latencies = []
for bucket in data['buckets']:
    if bucket['total']['count'] > 0:
        avg_duration = bucket['total']['duration'] / bucket['total']['count']
        latencies.append(avg_duration)
        hour = bucket['timestamp'][11:16]
        print(f"  {hour}: {avg_duration:.0f}ms")

if latencies:
    print(f"\nStatistics:")
    print(f"  Min: {min(latencies):.0f}ms")
    print(f"  Max: {max(latencies):.0f}ms")
    print(f"  Avg: {sum(latencies)/len(latencies):.0f}ms")

Filter by Successful Traces Only

Analyze only successful requests:

# Get metrics for successful traces only
newest = datetime.now(timezone.utc)
oldest = newest - timedelta(days=7)

payload = {
    "focus": "trace",
    "interval": 1440,
    "windowing": {
        "oldest": oldest.isoformat(),
        "newest": newest.isoformat()
    },
    "filter": {
        "conditions": [
            {
                "field": "status.code",
                "operator": "eq",
                "value": "STATUS_CODE_OK"
            }
        ]
    }
}

response = requests.post(BASE_URL, headers=headers, json=payload)
data = response.json()

# Calculate success metrics
total_count = sum(b['total']['count'] for b in data['buckets'])
total_cost = sum(b['total']['costs'] for b in data['buckets'])
total_duration = sum(b['total']['duration'] for b in data['buckets'])

print("Successful Traces (Last 7 Days):")
print(f"  Count: {total_count:,}")
print(f"  Total Cost: ${total_cost:.4f}")
print(f"  Avg Duration: {total_duration/total_count:.0f}ms")

Next Steps

• Learn about Query API for detailed trace data
• Explore Using the UI for visual analytics
• Read about Semantic Conventions for available metrics