Tracing de agentes LLM com OpenTelemetry: spans, tokens, custo e privacidade

python -m venv .venv
source .venv/bin/activate   # on Windows: .venv\Scripts\activate
pip install "opentelemetry-api==1.44.0" "opentelemetry-sdk==1.44.0"

"""A deterministic stand-in for a model provider and three backend tools."""
 
import itertools
import random
import re
import time
from dataclasses import dataclass, field
 
SYSTEM_PROMPT = "You are the support agent for an online store. Use tools for order data. Never invent order ids."
SYSTEM_TOKENS = 850  # instructions plus tool definitions, as the provider would count them
 
 
class RateLimited(Exception):
    status_code = 429
 
 
class ToolTimeout(Exception):
    pass
 
 
@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict
 
 
@dataclass
class ModelResponse:
    id: str
    model: str
    text: str | None
    tool_calls: list[ToolCall] = field(default_factory=list)
    finish_reason: str = "stop"
    input_tokens: int = 0
    output_tokens: int = 0
    cache_read_tokens: int = 0
 
 
def _tokens(text: str) -> int:
    return max(1, len(text) // 4)
 
 
class FakeModel:
    """Answers by rule instead of by sampling, and reports usage like a real API."""
 
    LATENCY = {"orion-large": (0.040, 0.0005), "orion-mini": (0.015, 0.0002)}
 
    def __init__(self, seed: int = 7):
        self._rng = random.Random(seed)
        self._ids = itertools.count(1)
        self._calls = 0
 
    def complete(self, model: str, messages: list[dict]) -> ModelResponse:
        self._calls += 1
        if self._calls % 30 == 0:
            time.sleep(0.005)
            raise RateLimited("429 Too Many Requests")
 
        last = messages[-1]
        prompt_tokens = SYSTEM_TOKENS + sum(_tokens(str(m["content"])) for m in messages)
        cached = SYSTEM_TOKENS if len(messages) > 1 else 0
        response_id = f"resp_{next(self._ids):05d}"
 
        if last["role"] == "user" and (match := re.search(r"ORD-\d{4}", last["content"])):
            tool = "lookup_tracking" if "where" in last["content"].lower() else "get_order"
            call = ToolCall(f"call_{response_id[5:]}", tool, {"order_id": match.group()})
            response = ModelResponse(response_id, model, None, [call], "tool_call", prompt_tokens, 24, cached)
        else:
            text = self._answer(last)
            response = ModelResponse(response_id, model, text, [], "stop", prompt_tokens,
                                     _tokens(text) + self._rng.randint(60, 220), cached)
 
        base, per_token = self.LATENCY[model]
        congestion = 0.4 if self._rng.random() < 0.04 else 0.0
        time.sleep(base + per_token * response.output_tokens + congestion)
        return response
 
    @staticmethod
    def _answer(last: dict) -> str:
        if last["role"] == "tool" and "error" in last["content"]:
            return "I couldn't reach the carrier right now. I've logged it, please try again in a few minutes."
        if last["role"] == "tool":
            return f"Here is what I found: {last['content']}"
        return "Summary: customer reports a delayed delivery and asks for a status update."
 
 
class Tools:
    def lookup_tracking(self, order_id: str, attempt: int) -> dict:
        number = int(order_id[4:])
        time.sleep(0.012)
        if number % 6 == 0 and (attempt == 1 or number % 12 == 0):
            time.sleep(0.2)
            raise ToolTimeout(f"carrier API timed out for {order_id}")
        return {"order_id": order_id, "status": "in_transit", "eta": "2026-09-21"}
 
    def get_order(self, order_id: str, attempt: int) -> dict:
        time.sleep(0.006)
        return {"order_id": order_id, "total": "189.90", "currency": "BRL", "refundable": True}
 
    def search_kb(self, query: str, top_k: int) -> list[dict]:
        time.sleep(0.010)
        policy = "Refunds are accepted within 30 days of delivery for unused items in the original packaging. " * 6
        return [{"id": f"kb-{i}", "score": round(0.9 - i * 0.07, 2), "text": policy} for i in range(top_k)]

import json
import re
from typing import Sequence
 
from opentelemetry import baggage, trace
from opentelemetry._logs import set_logger_provider
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import LogRecordExporter, LogRecordExportResult, SimpleLogRecordProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import Event, ReadableSpan, SpanProcessor, TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
 
# Business context that every span should carry, copied from baggage at span start.
CONTEXT_KEYS = ("app.tenant.id", "app.feature")
 
# Attributes that may hold user content. Everything else is operational metadata.
CONTENT_KEYS = {
    "gen_ai.input.messages",
    "gen_ai.output.messages",
    "gen_ai.system_instructions",
    "gen_ai.tool.call.arguments",
    "gen_ai.tool.call.result",
    "gen_ai.retrieval.query.text",
}
 
PII_PATTERNS = [
    (re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+"), "<email>"),
    (re.compile(r"\b(?:\d[ -]?){13,16}\b"), "<card>"),
    (re.compile(r"\b\d{3}\.\d{3}\.\d{3}-\d{2}\b"), "<cpf>"),
    (re.compile(r"\+?\d{2}\s?\(?\d{2}\)?\s?9?\d{4}-?\d{4}\b"), "<phone>"),
]
MAX_TEXT_CHARS = 300
 
 
def redact_text(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    if len(text) > MAX_TEXT_CHARS:
        text = text[:MAX_TEXT_CHARS] + f"...<{len(text) - MAX_TEXT_CHARS} chars truncated>"
    return text
 
 
def scrub(value):
    if isinstance(value, str):
        return redact_text(value)
    if isinstance(value, list):
        return [scrub(v) for v in value]
    if isinstance(value, dict):
        return {k: scrub(v) for k, v in value.items()}
    return value
 
 
def redact(raw: str) -> str:
    # Content attributes are JSON strings; scrub the strings inside and keep the structure valid.
    try:
        return json.dumps(scrub(json.loads(raw)))
    except ValueError:
        return redact_text(raw)
 
 
class BaggageAttributesProcessor(SpanProcessor):
    """Stamps tenant and feature on every span, so any span can be grouped without a join."""
 
    def on_start(self, span, parent_context=None):
        for key in CONTEXT_KEYS:
            value = baggage.get_baggage(key, parent_context)
            if value is not None:
                span.set_attribute(key, str(value))
 
 
class RedactingSpanProcessor(SpanProcessor):
    """Scrubs content attributes, then hands a sanitized copy to the exporting processor.
 
    A finished span is read-only in the SDK, so redaction happens on a copy.
    Wrapping the exporter's processor makes the order explicit: nothing reaches
    the exporter without passing through here first.
    """
 
    def __init__(self, delegate: SpanProcessor):
        self._delegate = delegate
 
    def on_start(self, span, parent_context=None):
        self._delegate.on_start(span, parent_context)
 
    def on_end(self, span: ReadableSpan):
        self._delegate.on_end(
            ReadableSpan(
                name=span.name,
                context=span.context,
                parent=span.parent,
                resource=span.resource,
                attributes=self._clean(span.attributes),
                events=[Event(ev.name, self._clean(ev.attributes), ev.timestamp) for ev in span.events],
                links=span.links,
                kind=span.kind,
                status=span.status,
                start_time=span.start_time,
                end_time=span.end_time,
                instrumentation_scope=span.instrumentation_scope,
            )
        )
 
    @staticmethod
    def _clean(attributes):
        cleaned = dict(attributes or {})
        for key, value in cleaned.items():
            if key in CONTENT_KEYS and isinstance(value, str):
                cleaned[key] = redact(value)
        return cleaned
 
    def shutdown(self):
        self._delegate.shutdown()
 
    def force_flush(self, timeout_millis: int = 30_000) -> bool:
        return self._delegate.force_flush(timeout_millis)
 
 
class JsonLinesSpanExporter(SpanExporter):
    def __init__(self, path: str):
        self._file = open(path, "w", encoding="utf-8")
 
    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        for span in spans:
            self._file.write(span.to_json(indent=None) + "\n")
        return SpanExportResult.SUCCESS
 
    def shutdown(self):
        self._file.close()
 
 
class JsonLinesLogExporter(LogRecordExporter):
    def __init__(self, path: str):
        self._file = open(path, "w", encoding="utf-8")
 
    def export(self, batch) -> LogRecordExportResult:
        for record in batch:
            self._file.write(record.to_json(indent=None) + "\n")
        return LogRecordExportResult.SUCCESS
 
    def shutdown(self):
        self._file.close()
 
    def force_flush(self, timeout_millis: int = 10_000) -> bool:
        self._file.flush()
        return True
 
 
def setup(service_name: str, spans_path: str, logs_path: str) -> tuple[TracerProvider, LoggerProvider]:
    resource = Resource.create({"service.name": service_name, "deployment.environment.name": "dev"})
 
    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(BaggageAttributesProcessor())
    # SimpleSpanProcessor keeps the demo synchronous; production uses BatchSpanProcessor.
    exporting = SimpleSpanProcessor(JsonLinesSpanExporter(spans_path))
    tracer_provider.add_span_processor(RedactingSpanProcessor(exporting))
    trace.set_tracer_provider(tracer_provider)
 
    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(SimpleLogRecordProcessor(JsonLinesLogExporter(logs_path)))
    set_logger_provider(logger_provider)
    return tracer_provider, logger_provider

import json
import os
import time
 
from opentelemetry import baggage, context, trace
from opentelemetry._logs import get_logger
from opentelemetry.trace import SpanKind, Status, StatusCode
 
import telemetry
from fakes import SYSTEM_PROMPT, FakeModel, RateLimited, Tools, ToolTimeout
 
AGENT = "support-agent"
PROVIDER = "fake_llm"
MODEL_BY_FEATURE = {"order_status": "orion-mini", "refund_help": "orion-large", "summarize_ticket": "orion-mini"}
CAPTURE_CONTENT = os.getenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "").upper() in ("SPAN_ONLY", "SPAN_AND_EVENT")
 
tracer = trace.get_tracer("support-agent", "1.0.0")
events = get_logger("support-agent.events")
llm = FakeModel()
tools = Tools()
 
 
def otel_messages(messages: list[dict]) -> str:
    parts = []
    for m in messages:
        if m["role"] == "tool":
            parts.append({"role": "tool", "parts": [{"type": "tool_call_response", "id": m["tool_call_id"], "response": m["content"]}]})
        elif m.get("tool_calls"):
            parts.append({"role": "assistant", "parts": [
                {"type": "tool_call", "id": c.id, "name": c.name, "arguments": c.arguments} for c in m["tool_calls"]]})
        else:
            parts.append({"role": m["role"], "parts": [{"type": "text", "content": m["content"]}]})
    return json.dumps(parts)
 
 
def chat(model: str, messages: list[dict]):
    attributes = {
        "gen_ai.operation.name": "chat",
        "gen_ai.provider.name": PROVIDER,
        "gen_ai.request.model": model,
        "gen_ai.request.temperature": 0.0,
        "gen_ai.request.max_tokens": 1024,
        "server.address": "llm.internal.example",
        "server.port": 443,
    }
    with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT, attributes=attributes) as span:
        if CAPTURE_CONTENT:
            span.set_attribute("gen_ai.system_instructions", json.dumps([{"type": "text", "content": SYSTEM_PROMPT}]))
            span.set_attribute("gen_ai.input.messages", otel_messages(messages))
        # One span per logical call: provider retries stay inside it, as events.
        for attempt in range(1, 4):
            try:
                response = llm.complete(model, messages)
                break
            except RateLimited:
                span.add_event("app.retry", {"error.type": "429", "app.retry.attempt": attempt})
                time.sleep(0.05 * attempt)
        else:
            span.set_attribute("error.type", "429")
            span.set_status(Status(StatusCode.ERROR, "rate limited after 3 attempts"))
            raise RateLimited("gave up")
 
        span.set_attributes({
            "gen_ai.response.id": response.id,
            "gen_ai.response.model": response.model,
            "gen_ai.response.finish_reasons": [response.finish_reason],
            "gen_ai.usage.input_tokens": response.input_tokens,
            "gen_ai.usage.output_tokens": response.output_tokens,
        })
        if response.cache_read_tokens:
            span.set_attribute("gen_ai.usage.cache_read.input_tokens", response.cache_read_tokens)
        if CAPTURE_CONTENT:
            parts = [{"type": "text", "content": response.text}] if response.text else [
                {"type": "tool_call", "id": c.id, "name": c.name, "arguments": c.arguments} for c in response.tool_calls]
            span.set_attribute("gen_ai.output.messages", json.dumps([{"role": "assistant", "parts": parts}]))
        return response, span
 
 
def execute_tool(call, attempt: int) -> dict:
    attributes = {
        "gen_ai.operation.name": "execute_tool",
        "gen_ai.agent.name": AGENT,
        "gen_ai.tool.name": call.name,
        "gen_ai.tool.call.id": call.id,
        "gen_ai.tool.type": "function",
    }
    with tracer.start_as_current_span(f"execute_tool {call.name}", kind=SpanKind.INTERNAL, attributes=attributes) as span:
        if CAPTURE_CONTENT:
            span.set_attribute("gen_ai.tool.call.arguments", json.dumps(call.arguments))
        try:
            result = getattr(tools, call.name)(call.arguments["order_id"], attempt)
        except ToolTimeout as exc:
            span.set_attribute("error.type", "timeout")
            span.set_status(Status(StatusCode.ERROR, str(exc)))
            return {"error": "timeout"}
        if CAPTURE_CONTENT:
            span.set_attribute("gen_ai.tool.call.result", json.dumps(result))
        return result
 
 
def retrieve(query: str, top_k: int = 4) -> list[dict]:
    attributes = {
        "gen_ai.operation.name": "retrieval",
        "gen_ai.data_source.id": "kb-refund-policies",
        "gen_ai.retrieval.top_k": top_k,
    }
    with tracer.start_as_current_span("retrieval kb-refund-policies", kind=SpanKind.CLIENT, attributes=attributes) as span:
        if CAPTURE_CONTENT:
            span.set_attribute("gen_ai.retrieval.query.text", query)
        return tools.search_kb(query, top_k)
 
 
def run_agent(conversation_id: str, feature: str, user_text: str) -> dict:
    model = MODEL_BY_FEATURE[feature]
    with tracer.start_as_current_span(f"invoke_agent {AGENT}", kind=SpanKind.INTERNAL, attributes={
        "gen_ai.operation.name": "invoke_agent",
        "gen_ai.agent.name": AGENT,
        "gen_ai.conversation.id": conversation_id,
        "gen_ai.request.model": model,
    }) as agent_span:
        content = user_text
        if feature == "refund_help":
            docs = retrieve(user_text)
            content += "\n\nPolicy excerpts:\n" + "\n".join(d["text"] for d in docs)
        messages = [{"role": "user", "content": content}]
        usage_in = usage_out = 0
        tool_failed = False
 
        for _ in range(4):  # step budget: an agent loop always needs one
            response, chat_span = chat(model, messages)
            usage_in += response.input_tokens
            usage_out += response.output_tokens
            if not response.tool_calls:
                break
            messages.append({"role": "assistant", "content": "", "tool_calls": response.tool_calls})
            for call in response.tool_calls:
                result = execute_tool(call, attempt=1)
                if "error" in result:
                    result = execute_tool(call, attempt=2)
                tool_failed = tool_failed or "error" in result
                messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)})
 
        agent_span.set_attributes({
            "gen_ai.usage.input_tokens": usage_in,
            "gen_ai.usage.output_tokens": usage_out,
            "gen_ai.response.finish_reasons": [response.finish_reason],
        })
        if tool_failed:
            agent_span.set_attribute("error.type", "tool_failed")
            agent_span.set_status(Status(StatusCode.ERROR, "answered without tool data"))
 
        # Stand-in for an online judge; a real one runs async, on a sample, and reports the same way.
        score = 2.0 if tool_failed else 4.0 + (int(conversation_id[-2:]) % 2)
        record_evaluation(chat_span, response.id, "Relevance", score)
        ids = agent_span.get_span_context()
        return {"trace_id": ids.trace_id, "span_id": ids.span_id, "tool_failed": tool_failed}
 
 
def record_evaluation(span, response_id: str, name: str, score: float) -> None:
    events.emit(
        event_name="gen_ai.evaluation.result",
        context=trace.set_span_in_context(span),  # parents the event to the chat span it grades
        attributes={
            "gen_ai.evaluation.name": name,
            "gen_ai.evaluation.score.value": score,
            "gen_ai.evaluation.score.label": "pass" if score >= 3 else "fail",
            "gen_ai.response.id": response_id,
        },
    )
 
 
def record_feedback(trace_id: int, span_id: int, rating: str) -> None:
    # Feedback arrives later, from another request: rebuild the span context from stored ids.
    parent = trace.NonRecordingSpan(trace.SpanContext(trace_id, span_id, is_remote=True,
                                                      trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED)))
    events.emit(event_name="app.user_feedback", context=trace.set_span_in_context(parent),
                attributes={"app.feedback.rating": rating})
 
 
REQUESTS = {
    "order_status": "Hi, I'm [email protected]. Where is my order ORD-{n}?",
    "refund_help": "Please refund order ORD-{n} to my card 4111 1111 1111 1111.",
    "summarize_ticket": "Summarize this ticket for the on-call agent: " + "The package was due Monday and has not arrived. " * 60,
}
TENANTS = ["acme", "acme", "globex", "acme", "initech", "globex"]  # acme sends half the traffic
 
 
def main() -> None:
    tracer_provider, logger_provider = telemetry.setup("support-agent", "spans.jsonl", "events.jsonl")
    features = ["order_status"] * 5 + ["refund_help"] * 3 + ["summarize_ticket"] * 2
    finished = []
    for i in range(120):
        feature = features[i % len(features)]
        tenant = TENANTS[i % len(TENANTS)]
        ctx = baggage.set_baggage("app.tenant.id", tenant)
        ctx = baggage.set_baggage("app.feature", feature, ctx)
        token = context.attach(ctx)
        try:
            finished.append(run_agent(f"conv_{i:04d}", feature, REQUESTS[feature].format(n=1000 + i)))
        finally:
            context.detach(token)
 
    for i, result in enumerate(finished):
        if result["tool_failed"] or i % 37 == 0:
            record_feedback(result["trace_id"], result["span_id"], "thumbs_down")
 
    tracer_provider.shutdown()
    logger_provider.shutdown()
    print(f"requests: {len(finished)}  content capture: {'on' if CAPTURE_CONTENT else 'off'}")
 
 
if __name__ == "__main__":
    main()

OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=SPAN_ONLY python agent.py

requests: 120  content capture: on

{
  "name": "chat orion-mini",
  "kind": "SpanKind.CLIENT",
  "parent_id": "0x97b82b18201929a1",
  "status": { "status_code": "UNSET" },
  "attributes": {
    "gen_ai.operation.name": "chat",
    "gen_ai.provider.name": "fake_llm",
    "gen_ai.request.model": "orion-mini",
    "gen_ai.request.temperature": 0.0,
    "gen_ai.request.max_tokens": 1024,
    "server.address": "llm.internal.example",
    "server.port": 443,
    "app.tenant.id": "globex",
    "app.feature": "order_status",
    "gen_ai.system_instructions": "[{\"type\": \"text\", \"content\": \"You are the support agent for an online store. Use tools for order data. Never invent order ids.\"}]",
    "gen_ai.input.messages": "[{\"role\": \"user\", \"parts\": [{\"type\": \"text\", \"content\": \"Hi, I'm <email>. Where is my order ORD-1002?\"}]}]",
    "gen_ai.response.id": "resp_00005",
    "gen_ai.response.model": "orion-mini",
    "gen_ai.response.finish_reasons": ["tool_call"],
    "gen_ai.usage.input_tokens": 865,
    "gen_ai.usage.output_tokens": 24,
    "gen_ai.output.messages": "[{\"role\": \"assistant\", \"parts\": [{\"type\": \"tool_call\", \"id\": \"call_00005\", \"name\": \"lookup_tracking\", \"arguments\": {\"order_id\": \"ORD-1002\"}}]}]"
  }
}
{
  "name": "execute_tool lookup_tracking",
  "kind": "SpanKind.INTERNAL",
  "parent_id": "0x97b82b18201929a1",
  "status": { "status_code": "ERROR", "description": "carrier API timed out for ORD-1002" },
  "attributes": {
    "gen_ai.operation.name": "execute_tool",
    "gen_ai.agent.name": "support-agent",
    "gen_ai.tool.name": "lookup_tracking",
    "gen_ai.tool.call.id": "call_00005",
    "gen_ai.tool.type": "function",
    "app.tenant.id": "globex",
    "app.feature": "order_status",
    "gen_ai.tool.call.arguments": "{\"order_id\": \"ORD-1002\"}",
    "error.type": "timeout"
  }
}
{
  "name": "invoke_agent support-agent",
  "kind": "SpanKind.INTERNAL",
  "parent_id": null,
  "status": { "status_code": "UNSET" },
  "attributes": {
    "gen_ai.operation.name": "invoke_agent",
    "gen_ai.agent.name": "support-agent",
    "gen_ai.conversation.id": "conv_0002",
    "gen_ai.request.model": "orion-mini",
    "app.tenant.id": "globex",
    "app.feature": "order_status",
    "gen_ai.usage.input_tokens": 1748,
    "gen_ai.usage.output_tokens": 235,
    "gen_ai.response.finish_reasons": ["stop"]
  }
}

"name": "chat orion-large",
"events": [{ "name": "app.retry", "attributes": { "error.type": "429", "app.retry.attempt": 1 } }]

import json
import math
from collections import defaultdict
from datetime import datetime
 
# USD per million tokens. Illustrative numbers for the fake models, versioned with the code.
PRICES = {
    "orion-large": {"input": 3.00, "cached_input": 0.30, "output": 15.00},
    "orion-mini": {"input": 0.25, "cached_input": 0.025, "output": 2.00},
}
 
 
def load(path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]
 
 
def seconds(span):
    start = datetime.fromisoformat(span["start_time"])
    end = datetime.fromisoformat(span["end_time"])
    return (end - start).total_seconds()
 
 
def p(values, q):
    ordered = sorted(values)
    return ordered[max(0, math.ceil(q * len(ordered)) - 1)]
 
 
def cost(attrs):
    price = PRICES[attrs["gen_ai.response.model"]]
    cached = attrs.get("gen_ai.usage.cache_read.input_tokens", 0)
    fresh = attrs["gen_ai.usage.input_tokens"] - cached
    return (fresh * price["input"] + cached * price["cached_input"]
            + attrs["gen_ai.usage.output_tokens"] * price["output"]) / 1_000_000
 
 
spans = load("spans.jsonl")
events = load("events.jsonl")
op = lambda s: s["attributes"].get("gen_ai.operation.name")
agents = [s for s in spans if op(s) == "invoke_agent"]
chats = [s for s in spans if op(s) == "chat"]
tool_calls = [s for s in spans if op(s) == "execute_tool"]
 
print("== cost and latency per feature")
by_feature = defaultdict(lambda: {"cost": 0.0, "tokens": 0, "latency": [], "requests": 0})
for s in chats:
    f = by_feature[s["attributes"]["app.feature"]]
    f["cost"] += cost(s["attributes"])
    f["tokens"] += s["attributes"]["gen_ai.usage.input_tokens"] + s["attributes"]["gen_ai.usage.output_tokens"]
for s in agents:
    f = by_feature[s["attributes"]["app.feature"]]
    f["requests"] += 1
    f["latency"].append(seconds(s))
print(f"{'feature':<18}{'requests':>9}{'cost USD':>11}{'USD/1k req':>12}{'tokens/req':>12}{'p95 s':>8}")
for name, f in sorted(by_feature.items(), key=lambda kv: -kv[1]["cost"]):
    print(f"{name:<18}{f['requests']:>9}{f['cost']:>11.4f}{1000 * f['cost'] / f['requests']:>12.2f}"
          f"{f['tokens'] // f['requests']:>12}{p(f['latency'], 0.95):>8.3f}")
 
print("\n== cost per tenant")
by_tenant = defaultdict(float)
for s in chats:
    by_tenant[s["attributes"]["app.tenant.id"]] += cost(s["attributes"])
for tenant, total in sorted(by_tenant.items(), key=lambda kv: -kv[1]):
    print(f"{tenant:<18}{total:>11.4f}")
 
print("\n== chat latency per model")
by_model = defaultdict(list)
for s in chats:
    by_model[s["attributes"]["gen_ai.request.model"]].append(seconds(s))
for model, values in sorted(by_model.items()):
    print(f"{model:<18}calls={len(values):<5}p50={p(values, 0.5):.3f}s  p95={p(values, 0.95):.3f}s  max={max(values):.3f}s")
retries = sum(1 for s in chats for ev in s["events"] if ev["name"] == "app.retry")
print(f"provider retries recorded as span events: {retries}")
 
print("\n== tools")
by_tool = defaultdict(lambda: [0, 0])
for s in tool_calls:
    t = by_tool[s["attributes"]["gen_ai.tool.name"]]
    t[0] += 1
    t[1] += s["status"]["status_code"] == "ERROR"
for tool, (calls, errors) in sorted(by_tool.items()):
    print(f"{tool:<18}calls={calls:<5}errors={errors:<4}error rate={errors / calls:.1%}")
failed = sum(1 for s in agents if s["attributes"].get("error.type") == "tool_failed")
print(f"requests answered without tool data: {failed} of {len(agents)}")
 
print("\n== quality signals joined to traces")
trace_of = {s["context"]["trace_id"]: s for s in agents}
scores = defaultdict(list)
for ev in events:
    if ev["event_name"] == "gen_ai.evaluation.result":
        root = trace_of[ev["trace_id"]]
        scores[root["attributes"]["app.feature"]].append(ev["attributes"]["gen_ai.evaluation.score.value"])
for feature, values in sorted(scores.items()):
    fails = sum(v < 3 for v in values)
    print(f"{feature:<18}relevance avg={sum(values) / len(values):.2f}  fail={fails}")
down = [ev for ev in events if ev["event_name"] == "app.user_feedback"]
with_errors = sum(1 for ev in down if trace_of[ev["trace_id"]]["status"]["status_code"] == "ERROR")
print(f"thumbs down: {len(down)}, of which {with_errors} land on traces with an error status")
 
print("\n== what a tail sampling policy would keep")
kept = [s for s in agents if s["status"]["status_code"] == "ERROR" or seconds(s) > 0.5
        or int(s["context"]["trace_id"], 16) % 10 == 0]
print(f"keep errors + slow (>500 ms) + 10% of the rest: {len(kept)} of {len(agents)} traces "
      f"({len(kept) / len(agents):.0%}), {sum(s['status']['status_code'] == 'ERROR' for s in kept)} errors kept")

python agent.py && python analyze.py

requests: 120  content capture: off
== cost and latency per feature
feature            requests   cost USD  USD/1k req  tokens/req   p95 s
refund_help              36     0.3296        9.16        3055   0.284
order_status             60     0.0361        0.60        1925   0.505
summarize_ticket         24     0.0169        0.71        1736   0.063
 
== cost per tenant
acme                   0.2075
globex                 0.1255
initech                0.0496
 
== chat latency per model
orion-large       calls=72   p50=0.091s  p95=0.160s  max=0.452s
orion-mini        calls=144  p50=0.037s  p95=0.098s  max=0.450s
provider retries recorded as span events: 7
 
== tools
get_order         calls=36   errors=0   error rate=0.0%
lookup_tracking   calls=72   errors=18  error rate=25.0%
requests answered without tool data: 6 of 120
 
== quality signals joined to traces
order_status      relevance avg=4.20  fail=6
refund_help       relevance avg=4.67  fail=0
summarize_ticket  relevance avg=4.50  fail=0
thumbs down: 10, of which 6 land on traces with an error status
 
== what a tail sampling policy would keep
keep errors + slow (>500 ms) + 10% of the rest: 26 of 120 traces (22%), 6 errors kept

# pip install opentelemetry-exporter-otlp-proto-grpc
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
exporting = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True))
tracer_provider.add_span_processor(RedactingSpanProcessor(exporting))

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
 
processors:
  redaction:
    allow_all_keys: true
    blocked_values:
      - "[\\w.+-]+@[\\w-]+(?:\\.[\\w-]+)+"
  tail_sampling:
    decision_wait: 30s
    policies:
      - name: errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: slow
        type: latency
        latency: { threshold_ms: 8000 }
      - name: baseline
        type: probabilistic
        probabilistic: { sampling_percentage: 10 }
  batch: {}
 
connectors:
  spanmetrics:
    dimensions:
      - name: gen_ai.operation.name
      - name: gen_ai.request.model
      - name: app.feature
 
exporters:
  otlp/tempo:
    endpoint: tempo:4317
    tls: { insecure: true }
  prometheusremotewrite:
    endpoint: http://prometheus:9090/api/v1/write
 
service:
  pipelines:
    traces/all:
      receivers: [otlp]
      processors: [redaction, batch]
      exporters: [spanmetrics]
    traces/sampled:
      receivers: [otlp]
      processors: [redaction, tail_sampling, batch]
      exporters: [otlp/tempo]
    metrics/spans:
      receivers: [spanmetrics]
      exporters: [prometheusremotewrite]