In production AI systems, understanding where your inference dollars go is the difference between a profitable product and a budget hemorrhage. When you're running thousands of API calls per minute across multiple models—GPT-4.1 at $8/MTok, Claude Sonnet 4.5 at $15/MTok, or budget options like DeepSeek V3.2 at just $0.42/MTok—the ability to trace request chains, identify cost bottlenecks, and optimize token usage becomes mission-critical.
This guide dives deep into implementing OpenTelemetry for comprehensive AI API observability, covering architecture design, high-performance tracing, and production-grade patterns that handle millions of spans daily.
Why OpenTelemetry for AI APIs?
Traditional logging tells you what happened. OpenTelemetry tells you why it cost what it did. When you integrate tracing at the API layer, you gain:
- Token-level cost attribution: Map every span to input/output token counts
- Latency breakdowns: Identify whether delays come from model inference, network transit, or serialization
- Multi-model correlation: Trace requests that fan out across GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, and DeepSeek V3.2
- Anomaly detection: Spot sudden cost spikes before they destroy your monthly budget
Architecture Overview
The tracing architecture consists of four core components working in concert:
+---------------------+ +----------------------+ +------------------+
| Your Application |----▶| OTel Collector |----▶| Jaeger/Tempo |
| (Instrumented SDK) | | (Batch Processor) | | (Trace Storage) |
+---------------------+ +----------------------+ +------------------+
│ │
▼ ▼
+------------------+ +------------------+
| HolySheep AI | | Cost Dashboard |
| api.holysheep.ai| | (Prometheus) |
+------------------+ +------------------+
Core Implementation: Tracing Client with Cost Attribution
Here's a production-grade Python implementation that wraps the HolySheep AI API with OpenTelemetry instrumentation, automatically capturing token counts and latency for every request.
import asyncio
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.trace import Status, StatusCode
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from dataclasses import dataclass
from typing import Optional, Dict, Any
import aiohttp
import time
import json
Initialize OpenTelemetry with resource metadata
resource = Resource.create({
"service.name": "ai-cost-tracer",
"service.version": "1.0.0",
"deployment.environment": "production"
})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
@dataclass
class TokenUsage:
"""Structured token usage data for cost attribution."""
input_tokens: int
output_tokens: int
model: str
cost_usd: float
class HolySheepTracedClient:
"""
Production-grade traced client for HolySheep AI API.
Automatically captures token usage and latency for cost analysis.
"""
PRICING = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 0.35, "output": 2.50},
"deepseek-v3.2": {"input": 0.14, "output": 0.42},
}
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=60, connect=10)
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=timeout
)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Calculate cost in USD based on 2026 pricing."""
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6)
async def chat_completion(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: Optional[int] = None,
trace_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Execute traced chat completion with automatic cost attribution.
"""
span_name = f"ai.chat.{model}"
with tracer.start_as_current_span(span_name) as span:
# Set span attributes for filtering and grouping
span.set_attribute("ai.model", model)
span.set_attribute("ai.temperature", temperature)
span.set_attribute("ai.max_tokens", max_tokens or 0)
span.set_attribute("ai.provider", "holysheep")
# Estimate input tokens (approximate for span creation)
estimated_input = sum(len(str(m)) for m in messages) // 4
span.set_attribute("ai.input_tokens.estimated", estimated_input)
start_time = time.perf_counter()
try:
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
async with self._session.post(
f"{self.base_url}/chat/completions",
json=payload
) as response:
response.raise_for_status()
data = await response.json()
# Extract actual token usage from response
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", estimated_input)
output_tokens = usage.get("completion_tokens", 0)
# Calculate and record cost
cost = self._calculate_cost(model, input_tokens, output_tokens)
latency_ms = (time.perf_counter() - start_time) * 1000
# Set comprehensive span attributes
span.set_attribute("ai.input_tokens", input_tokens)
span.set_attribute("ai.output_tokens", output_tokens)
span.set_attribute("ai.total_tokens", input_tokens + output_tokens)
span.set_attribute("ai.cost_usd", cost)
span.set_attribute("ai.latency_ms", round(latency_ms, 2))
span.set_attribute("ai.response_id", data.get("id", ""))
# Add cost to span as metric annotation
span.add_event(
"token_usage",
{
"input": input_tokens,
"output": output_tokens,
"cost_usd": cost
}
)
span.set_status(Status(StatusCode.OK))
return {
**data,
"_trace": {
"cost_usd": cost,
"latency_ms": latency_ms,
"input_tokens": input_tokens,
"output_tokens": output_tokens
}
}
except Exception as e:
span.set_status(Status(StatusCode.ERROR, str(e)))
span.record_exception(e)
raise
async def example_multi_model_trace():
"""Demonstrate tracing across multiple models with cost comparison."""
client = HolySheepTracedClient(api_key="YOUR_HOLYSHEEP_API_KEY")
async with client:
prompt = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain the concept of distributed tracing in one paragraph."}
]
# Trace calls to different models
models = ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1", "claude-sonnet-4.5"]
print("Model Cost Comparison (2026 Pricing):")
print("-" * 60)
for model in models:
result = await client.chat_completion(model=model, messages=prompt)
trace_info = result["_trace"]
print(f"{model:25} | "
f"Tokens: {trace_info['input_tokens']:5}+{trace_info['output_tokens']:4} | "
f"Cost: ${trace_info['cost_usd']:.6f} | "
f"Latency: {trace_info['latency_ms']:.1f}ms")
asyncio.run(example_multi_model_trace())
High-Performance Async Batch Processing
For production workloads handling thousands of requests, batch processing with semaphore-controlled concurrency prevents rate limit errors while maximizing throughput.
import asyncio
from typing import List, Dict, Any, Callable
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class BatchRequest:
"""Encapsulates a batch request with metadata for tracing."""
request_id: str
model: str
messages: list
temperature: float = 0.7
max_tokens: Optional[int] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class BatchResult:
"""Aggregated results from batch processing."""
total_requests: int
successful: int
failed: int
total_input_tokens: int
total_output_tokens: int
total_cost_usd: float
total_latency_ms: float
cost_by_model: Dict[str, float]
latency_p50_ms: float
latency_p95_ms: float
latency_p99_ms: float
class TracedBatchProcessor:
"""
High-performance batch processor with concurrency control.
Handles rate limiting automatically and provides detailed cost analytics.
"""
def __init__(
self,
client: HolySheepTracedClient,
max_concurrent: int = 10,
rate_limit_rpm: int = 500
):
self.client = client
self.semaphore = asyncio.Semaphore(max_concurrent)
self.rate_limiter = RateLimiter(rate_limit_rpm)
self._metrics: List[float] = []
async def process_batch(
self,
requests: List[BatchRequest],
progress_callback: Optional[Callable[[int, int], None]] = None
) -> BatchResult:
"""Process a batch of requests with automatic concurrency and rate limiting."""
start_time = time.perf_counter()
tasks = []
for i, req in enumerate(requests):
task = self._process_single(req, i, progress_callback)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# Aggregate metrics
total_input = total_output = total_cost = 0
cost_by_model = defaultdict(float)
latencies = []
successful = failed = 0
for result in results:
if isinstance(result, Exception):
failed += 1
else:
successful += 1
total_input += result["input_tokens"]
total_output += result["output_tokens"]
total_cost += result["cost_usd"]
cost_by_model[result["model"]] += result["cost_usd"]
latencies.append(result["latency_ms"])
latencies.sort()
n = len(latencies)
return BatchResult(
total_requests=len(requests),
successful=successful,
failed=failed,
total_input_tokens=total_input,
total_output_tokens=total_output,
total_cost_usd=round(total_cost, 6),
total_latency_ms=(time.perf_counter() - start_time) * 1000,
cost_by_model=dict(cost_by_model),
latency_p50_ms=latencies[int(n * 0.5)] if n else 0,
latency_p95_ms=latencies[int(n * 0.95)] if n else 0,
latency_p99_ms=latencies[int(n * 0.99)] if n else 0,
)
async def _process_single(
self,
req: BatchRequest,
index: int,
callback: Optional[Callable]
) -> Dict[str, Any]:
"""Process a single request with rate limiting and concurrency control."""
async with self.semaphore:
await self.rate_limiter.acquire()
if callback:
callback(index + 1, len(await self._get_pending_count()))
try:
result = await self.client.chat_completion(
model=req.model,
messages=req.messages,
temperature=req.temperature,
max_tokens=req.max_tokens
)
trace_info = result["_trace"]
self._metrics.append(trace_info["latency_ms"])
return {
"request_id": req.request_id,
"model": req.model,
"input_tokens": trace_info["input_tokens"],
"output_tokens": trace_info["output_tokens"],
"cost_usd": trace_info["cost_usd"],
"latency_ms": trace_info["latency_ms"],
"success": True
}
except Exception as e:
return {
"request_id": req.request_id,
"model": req.model,
"error": str(e),
"success": False
}
class RateLimiter:
"""Token bucket rate limiter for API calls."""
def __init__(self, requests_per_minute: int):
self.rpm = requests_per_minute
self.interval = 60.0 / requests_per_minute
self.last_call = 0.0
self._lock = asyncio.Lock()
async def acquire(self):
async with self._lock:
now = time.monotonic()
wait_time = max(0, self.interval - (now - self.last_call))
if wait_time > 0:
await asyncio.sleep(wait_time)
self.last_call = time.monotonic()
Benchmark: Process 1000 requests across multiple models
async def benchmark_batch_processing():
"""Run performance benchmark on batch processing."""
client = HolySheepTracedClient(api_key="YOUR_HOLYSHEEP_API_KEY")
processor = TracedBatchProcessor(
client=client,
max_concurrent=15,
rate_limit_rpm=500
)
# Generate test requests
test_requests = [
BatchRequest(
request_id=f"req_{i}",
model=["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"][i % 3],
messages=[
{"role": "user", "content": f"Request {i}: Tell me about distributed systems."}
]
)
for i in range(1000)
]
print("Starting batch benchmark: 1000 requests")
start = time.perf_counter()
result = await processor.process_batch(test_requests)
elapsed = time.perf_counter() - start
print(f"\nBenchmark Results:")
print(f" Total Requests: {result.total_requests}")
print(f" Successful: {result.successful}")
print(f" Failed: {result.failed}")
print(f" Total Cost: ${result.total_cost_usd:.4f}")
print(f" Throughput: {result.total_requests / elapsed:.1f} req/s")
print(f" P50 Latency: {result.latency_p50_ms:.1f}ms")
print(f" P95 Latency: {result.latency_p95_ms:.1f}ms")
print(f" P99 Latency: {result.latency_p99_ms:.1f}ms")
print(f"\nCost by Model:")
for model, cost in result.cost_by_model.items():
print(f" {model}: ${cost:.4f}")
asyncio.run(benchmark_batch_processing())