Structured traces

Unstructured logs are where debugging dies. You grep for a thread id, find 40 unrelated lines, and still cannot tell what actually happened. Structured traces fix that by giving every layer the same shape: a name, a status, a duration, optional notes, optional data. Parse once, query anywhere.

layers/observability.py
python
LAYER_NAMES = [
    "transport",
    "orchestrator",
    "tools",
    "memory",
    "retrieval",
    "guardrails",
    "observability",
]


class ObservabilityLayer:
    def __init__(self) -> None:
        self._last_trace: Dict[str, List[dict]] = {}
        self._log = structlog.get_logger("production-agent")

    def start_span(self) -> float:
        return time.perf_counter()

    def record(
        self,
        trace: List[dict],
        layer: str,
        start: float,
        status: str = "ok",
        notes: Optional[str] = None,
        data: Optional[dict] = None,
    ) -> None:
        duration_ms = (time.perf_counter() - start) * 1000.0
        trace.append({
            "layer": layer,
            "status": status,
            "duration_ms": round(duration_ms, 3),
            "notes": notes,
            "data": data,
        })

Every entry is the same shape. You can pipe traces into OpenTelemetry, ship them to your log aggregator, or replay them locally. The record function is the single choke point for trace writes.

service.py
python
async def run(self, message: str, thread_id: str) -> ChatResponse:
    trace: List[dict] = []
    obs = self.observability

    # Layer 1: Transport
    t0 = obs.start_span()
    normalized = self.transport.validate(message, thread_id)
    obs.record(trace, "transport", t0, notes="validated")

    # Layer 6a: Guardrails (input)
    t0 = obs.start_span()
    gi = self.guardrails.check_input(normalized["message"])
    obs.record(trace, "guardrails", t0, notes="input", data={"flags": gi.flags})
    clean_message = gi.text

    # Layer 4a: Memory read
    t0 = obs.start_span()
    history = self.memory.history(thread_id)
    obs.record(trace, "memory", t0, notes="read", data={"turns": len(history)})

    # Layer 2: Orchestrator
    t0 = obs.start_span()
    state = self.orchestrator.initial_state(clean_message, thread_id)
    route = self.orchestrator.route(state)
    obs.record(trace, "orchestrator", t0, notes=f"route={route}")

The service wires every layer call between a start_span and a record. The trace is the only thing the top-level run() builds up; each layer remains ignorant of observability details.

Quiz: Quiz

Loading practice…