AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Implement distributed tracing spans for LLM calls

Implement Distributed Tracing Spans for LLM Calls

Standard application tracing shows HTTP latency. For LLM pipelines you need per-span visibility: retrieval, prompt assembly, model call, and postprocessing — each with its own latency and metadata.

What to build

A lightweight tracing system (no external dependencies):

Span dataclass: trace_id, span_id, parent_id, name, start_time, end_time, attributes: dict, status ("ok"/"error"), error_message. Include duration_ms property.
Tracer:
- start_span(name, parent=None) -> Span — child spans inherit parent.trace_id
- finish_span(span, status="ok", error=None)
- add_attribute(span, key, value)
- get_trace(trace_id) -> list[Span] — sorted by start_time
- export_trace(trace_id) -> list[dict] — serializable dicts with duration_ms
trace_span(tracer, name, parent=None) context manager — starts, yields, finishes (error status on exception).

Why it matters

Without spans, "the request took 3 seconds" tells you nothing about where time was spent. Spans reveal whether the bottleneck was retrieval, prompt assembly, or the model call.

Api Async / medium / Step 17 of 23

Practice stage

Async and provider control

Hint

Make waiting behavior explicit. Timeouts, retries, and concurrency limits matter more than squeezing everything into one helper.

Next drill

Implement graceful degradation with multi-level fallback tiers

Success criteria

- Uses async boundaries coherently
- Makes timeout and retry decisions legible
- Would be maintainable under provider instability

Review checklist

- Is timeout behavior explicit?
- Is retryable failure separate from terminal failure?
- Would logs reveal what actually timed out?

from __future__ import annotations
import time
import uuid
from collections import defaultdict
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Generator

@dataclass
class Span:
    trace_id: str
    span_id: str
    name: str
    start_time: float
    parent_id: str | None = None
    end_time: float | None = None
    attributes: dict = field(default_factory=dict)
    status: str = "ok"
    error_message: str | None = None

@property
    def duration_ms(self) -> float | None:
        if self.end_time is None:
            return None
        return round((self.end_time - self.start_time) * 1000, 2)

class Tracer:
    def __init__(self):
        raise NotImplementedError

def start_span(self, name: str, parent: Span | None = None) -> Span:
        raise NotImplementedError

def finish_span(self, span: Span, status: str = "ok", error: str | None = None) -> None:
        raise NotImplementedError

def add_attribute(self, span: Span, key: str, value) -> None:
        raise NotImplementedError

def get_trace(self, trace_id: str) -> list[Span]:
        raise NotImplementedError

def export_trace(self, trace_id: str) -> list[dict]:
        raise NotImplementedError

@contextmanager
def trace_span(tracer: Tracer, name: str, parent: Span | None = None) -> Generator[Span, None, None]:
    raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.