Observers

An observer receives hooks during a run: step start, cache hit, step finished, fan-out branch. Pass one with execute(observer=...).

• LoggingObserver: the default. Logs each step with structlog.
• NullObserver: silent. Subclass it and override only the hooks you need.
• RunTreeObserver: collects a timing tree; print it with render_tree().

"""Observers: watch a run. RunTreeObserver prints a timing tree with cache hits."""

import tempfile
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel

from adag import Pipeline, RunTreeObserver, Step


class Number(BaseModel):
    value: int


@dataclass(frozen=True)
class Load(Step[Number]):
    def run(self) -> Number:
        return Number(value=4)


@dataclass(frozen=True)
class Square(Step[Number]):
    def run(self, n: Number) -> Number:
        return Number(value=n.value**2)


graph = Square()(n=Load()())
cache = Path(tempfile.mkdtemp())

for label in ["first run", "second run"]:
    observer = RunTreeObserver()
    Pipeline(graph, cache_dir=cache).execute(observer=observer)
    print(f"{label}:\n{observer.render_tree()}")

first run:
Square  ● 0ms
└─ Load  ● 0ms
second run:
Square  ⟳ cached

The hooks are the methods of PipelineObserver: on_step_start, on_step_cached, on_step_finished, on_fan_out_branch, on_fan_out_branch_failed, and on_event for progress events (step.saved, batch.*, item_cache.*, branch.*).

• NullObserver drops all of them, so a run prints nothing.
• Warnings and errors (a timed-out branch, a cache invariant violation) are always logged with structlog, so an observer cannot hide a problem.
• Inside a step body, send your own event with adag.deferred.emit("my.event", n=3).

Typed hooks for your own values

Use a typed hook when your code consumes a value from a run (a score, a vector, a model name), instead of logging it. adag has no domain hooks of its own. Declare one yourself:

1. A runtime_checkable Protocol with your method, here on_score(score: float).
2. One function that calls the method when the observer has it, and does nothing otherwise.
3. An observer that implements the method. It subclasses NullObserver for the other hooks.

Call the function from either place:

• Inside a step, with context.observer (override execute, call super().execute).
• From any helper a step calls, with current_observer().

"""A typed domain hook: a step sends a value, a custom observer receives it."""

from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable

from pydantic import BaseModel

from adag import InputParam, NullObserver, Pipeline, Step
from adag.deferred import PipelineObserver, ResolveContext, current_observer


@runtime_checkable
class ScoreObserver(Protocol):
    """Declares the extra hook. Any observer that has this method receives scores."""

    def on_score(self, score: float) -> None: ...


def report_score(observer: PipelineObserver, score: float) -> None:
    """Call ``on_score`` when the observer has it; do nothing otherwise."""
    if isinstance(observer, ScoreObserver):
        observer.on_score(score)


def normalise(x: float) -> float:
    """A helper deep in the call stack: it reaches the observer of the current run."""
    score = x / 10
    report_score(current_observer(), score)
    return score


class Number(BaseModel):
    value: float


@dataclass(frozen=True)
class Score(Step[Number]):
    def run(self, x: float) -> Number:
        return Number(value=normalise(x))


@dataclass(frozen=True)
class Double(Step[Number]):
    def execute(self, context: ResolveContext, **inputs: object) -> Number:
        result = super().execute(context, **inputs)
        report_score(context.observer, result.value)  # a step with the context in hand
        return result

    def run(self, n: Number) -> Number:
        return Number(value=n.value * 2)


@dataclass
class CollectScores(NullObserver):
    scores: list[float] = field(default_factory=list[float])

    def on_score(self, score: float) -> None:
        self.scores.append(score)


pipeline = Pipeline(Double()(n=Score()(x=InputParam[float]("x"))))
observer = CollectScores()
pipeline.execute(x=7.0, observer=observer)
print(observer.scores)
pipeline.execute(x=7.0, observer=NullObserver())  # a plain observer: report_score does nothing

[0.7, 1.4]

(output trimmed: log lines removed)

A strict type checker sees float at both ends. A plain NullObserver or LoggingObserver keeps working, because the function skips an observer without the method.

on_event(event, fields) stays untyped (dict[str, Any]): it carries progress events, not values a caller consumes.

The deprecated NullObserver.on_embedding and on_model_refused do nothing. Replace each with a typed hook as above.